From b7e757dbf748cbaa522d65eacbc6037721169c7b Mon Sep 17 00:00:00 2001 From: rustmailer Date: Mon, 13 Jul 2026 20:56:26 +0800 Subject: [PATCH] update --- Cargo.lock | 3 + crates/admin/Cargo.toml | 7 +- crates/admin/src/main.rs | 11 +- crates/admin/src/migrate_store_v2.rs | 520 +++++++++++++++++++++ crates/admin/src/migrate_v037.rs | 659 +++++++++++++++++++++++++++ crates/admin/src/migrate_v1.rs | 398 ++++++++++++++++ crates/blob/src/meta.rs | 23 +- crates/core/src/migrate/mod.rs | 8 +- crates/server/src/lib.rs | 6 +- 9 files changed, 1608 insertions(+), 27 deletions(-) create mode 100644 crates/admin/src/migrate_store_v2.rs create mode 100644 crates/admin/src/migrate_v037.rs create mode 100644 crates/admin/src/migrate_v1.rs diff --git a/Cargo.lock b/Cargo.lock index 5dfdff1..ceb1c61 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -301,6 +301,7 @@ checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" name = "bichon-admin" version = "1.6.2" dependencies = [ + "bichon-blob", "bichon-core", "bichon-memdb", "bytes 1.12.0", @@ -308,6 +309,7 @@ dependencies = [ "console", "dialoguer", "fjall", + "hex", "indicatif", "itertools 0.15.0", "mail-parser", @@ -317,6 +319,7 @@ dependencies = [ "serde_json", "snafu", "tantivy", + "tempfile", "tokio", "tracing", "uuid", diff --git a/crates/admin/Cargo.toml b/crates/admin/Cargo.toml index 02f34d1..216ecdd 100644 --- a/crates/admin/Cargo.toml +++ b/crates/admin/Cargo.toml @@ -19,9 +19,14 @@ itertools.workspace = true snafu.workspace = true bichon-memdb.workspace = true fjall.workspace = true +hex.workspace = true +bichon-blob.workspace = true mail-parser.workspace = true bytes.workspace = true uuid.workspace = true tantivy = { version = "0.26.1", features = ["zstd-compression", "quickwit"] } chrono.workspace = true -tracing.workspace = true \ No newline at end of file +tracing.workspace = true + +[dev-dependencies] +tempfile = "3" \ No newline at end of file diff --git a/crates/admin/src/main.rs b/crates/admin/src/main.rs index 8f95a00..0fa42cd 100644 --- a/crates/admin/src/main.rs +++ b/crates/admin/src/main.rs @@ -19,12 +19,15 @@ use console::style; use dialoguer::{theme::ColorfulTheme, Select}; -use crate::{migrate::handle_migration, reset::handle_reset_password}; +use crate::{migrate::handle_migration, 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; pub mod reset; @@ -42,7 +45,9 @@ async fn run_interactive() { let main_options = vec![ "Reset Admin Password", - "Migrate Legacy v0.3.7 Storage to v1.x", + "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", ]; @@ -56,6 +61,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), _ => { println!("{}", style("Exiting...").dim()); } diff --git a/crates/admin/src/migrate_store_v2.rs b/crates/admin/src/migrate_store_v2.rs new file mode 100644 index 0000000..3881676 --- /dev/null +++ b/crates/admin/src/migrate_store_v2.rs @@ -0,0 +1,520 @@ +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 bichon_blob::{Codec, Config, Engine}; +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)>, +} + +fn hex_to_raw_key(hex: &str) -> BichonResult<[u8; 32]> { + let mut key = [0u8; 32]; + hex::decode_to_slice(hex, &mut key).map_err(|e| { + raise_error!( + format!("invalid content hash: {e:#?}"), + ErrorCode::InternalError + ) + })?; + Ok(key) +} + +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 NewIndexWriterV2 { + pub envelope_writer: Option, + pub attachment_writer: Option, + pub engine: Engine, + pending: usize, + email_buf: Vec<([u8; 32], Vec)>, + attachment_buf: Vec<([u8; 32], Vec)>, +} + +impl NewIndexWriterV2 { + 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))?; + + 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))?; + + attachment_writer.set_merge_policy(Box::new(NoMergePolicy)); + + // ── blob store (bichon-blob, not fjall) ─────────────────────────── + std::fs::create_dir_all(&dirs.storage_dir) + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + let blob_dir = dirs.storage_dir.join("blobs"); + + let mut config = Config::default(); + config.default_codec = Codec::Zstd; + config.compress_threshold = 1024; + config.flush_interval_secs = 0; + config.gc_interval_secs = 0; + + let engine = Engine::open(&blob_dir, config) + .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; + + Ok(Self { + envelope_writer: Some(envelope_writer), + attachment_writer: Some(attachment_writer), + engine, + 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 email_raw_key = hex_to_raw_key(&email_content_hash)?; + + 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 write — sorted + flushed later. + // Key is the raw 32-byte hash (not the hex string). + self.email_buf + .push((email_raw_key, stripped_eml)); + for (hash, data) in &attachment_output.blobs { + let raw_key = hex_to_raw_key(hash)?; + self.attachment_buf.push((raw_key, 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; + + 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(()) + } + + /// Write buffered blobs to the bichon-blob engine. + /// Also commits the Tantivy writers to bound their in-memory state. + pub fn flush_blob_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 count = self.email_buf.len(); + for (key, data) in &self.email_buf { + self.engine.put(*key, data, Codec::Zstd).map_err(|e| { + raise_error!( + format!("blob engine put error: {e:#?}"), + ErrorCode::InternalError + ) + })?; + } + println!("flushed {} email blobs to engine", count); + 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 count = self.attachment_buf.len(); + for (key, data) in &self.attachment_buf { + self.engine.put(*key, data, Codec::Zstd).map_err(|e| { + raise_error!( + format!("blob engine put error: {e:#?}"), + ErrorCode::InternalError + ) + })?; + } + println!("flushed {} attachment blobs to engine", count); + self.attachment_buf.clear(); + } + + Ok(()) + } + + /// Flush and shutdown the blob engine (called once at the very end). + pub fn shutdown_engine(&mut self) -> BichonResult<()> { + self.engine.flush().map_err(|e| { + raise_error!( + format!("engine flush error: {e:#?}"), + ErrorCode::InternalError + ) + })?; + self.engine.shutdown().map_err(|e| { + raise_error!( + format!("engine shutdown error: {e:#?}"), + ErrorCode::InternalError + ) + })?; + Ok(()) + } +} diff --git a/crates/admin/src/migrate_v037.rs b/crates/admin/src/migrate_v037.rs new file mode 100644 index 0000000..cada749 --- /dev/null +++ b/crates/admin/src/migrate_v037.rs @@ -0,0 +1,659 @@ +use std::{collections::HashMap, path::PathBuf}; + +use bichon_core::{ + error::{code::ErrorCode, BichonResult}, + migrate::write_storage_version, + 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::is_legacy_data_layout_with_paths; +use crate::migrate_store::LegacyDirs; +use crate::migrate_store_v2::{NewDirs, NewIndexWriterV2}; + +pub fn handle_migration_v037(theme: &ColorfulTheme) { + println!( + "\n{}", + style("MIGRATION: Bichon v0.3.7 Storage → v2.x (bichon-blob)") + .bold() + .yellow() + ); + + println!( + "{}", + style( + "This tool migrates data from the legacy v0.3.7 Tantivy-based storage \ + architecture directly to the v2.x bichon-blob 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 v2.x architecture:\n\ + • mail indexes stored in Tantivy\n\ + • attachment indexes stored in Tantivy\n\ + • raw message data stored in bichon-blob engine\n\ + • attachment blobs stored in bichon-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 v2.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 a newer 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 crate::migrate::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 NewIndexWriterV2::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_v2( + 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.set_message(style("Shutting down blob engine...").dim().to_string()); + if let Err(e) = writer.shutdown_engine() { + pb.finish_with_message(format!("{}", style("Migration failed.").red())); + eprintln!("\n{} {:?}", style("✘").red().bold(), e); + return; + } + + // Write STORAGE_VERSION = 2 to mark the data as v2.x compatible + if let Err(e) = write_storage_version(&root_path, 2) { + pb.finish_with_message(format!("{}", style("Migration failed.").red())); + eprintln!( + "\n{} Failed to write STORAGE_VERSION: {:?}", + 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 to v2.x completed successfully!").bold() + ); +} + +/// Migrate all documents from a single EML segment to the v2.x storage layout. +fn do_migrate_segment_v2( + batch_size: u32, + legacy: LegacyDirs, + writer: &mut NewIndexWriterV2, + 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); + + 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 to bichon-blob engine. + writer.flush_blob_buffers()?; + + chunk_start = chunk_end; + } + + on_progress(&format!("DONE:{}:{}", total_migrated, total_skipped)); + Ok(()) +} diff --git a/crates/admin/src/migrate_v1.rs b/crates/admin/src/migrate_v1.rs new file mode 100644 index 0000000..021565f --- /dev/null +++ b/crates/admin/src/migrate_v1.rs @@ -0,0 +1,398 @@ +use std::path::PathBuf; + +use bichon_blob::{Codec, Config, Engine}; +use bichon_core::{ + error::{code::ErrorCode, BichonResult}, + migrate::write_storage_version, + raise_error, +}; +use console::style; +use dialoguer::{theme::ColorfulTheme, Input}; +use fjall::{Config as FjallConfig, Database}; +use indicatif::{ProgressBar, ProgressStyle}; + +fn hex_key_to_raw(hex_bytes: &[u8]) -> BichonResult<[u8; 32]> { + let hex_str = std::str::from_utf8(hex_bytes).map_err(|e| { + raise_error!( + format!("invalid UTF-8 in fjall key: {e:#?}"), + ErrorCode::InternalError + ) + })?; + let mut raw = [0u8; 32]; + hex::decode_to_slice(hex_str, &mut raw).map_err(|e| { + raise_error!( + format!("invalid hex in fjall key '{hex_str}': {e:#?}"), + ErrorCode::InternalError + ) + })?; + Ok(raw) +} + +fn migrate_keyspace( + engine: &Engine, + db: &Database, + ks_name: &str, + label: &str, +) -> BichonResult { + let ks = db + .keyspace(ks_name, || { + panic!("{ks_name} keyspace not found in fjall database") + }) + .map_err(|e| { + raise_error!( + format!("failed to open fjall keyspace '{ks_name}': {e:#?}"), + ErrorCode::InternalError + ) + })?; + + let pb = ProgressBar::new_spinner(); + pb.set_style( + ProgressStyle::with_template("{spinner:.cyan} {msg} [{elapsed_precise}]") + .unwrap(), + ); + pb.set_message(format!("Scanning {label} blobs...")); + + let mut count: u64 = 0; + for item in ks.iter() { + let (key_bytes, value) = item.into_inner().map_err(|e| { + raise_error!( + format!("fjall iter error in '{ks_name}': {e:#?}"), + ErrorCode::InternalError + ) + })?; + + if value.is_empty() { + continue; + } + let raw_key = hex_key_to_raw(&key_bytes)?; + engine.put(raw_key, &value, Codec::Zstd).map_err(|e| { + raise_error!( + format!("bichon-blob put error: {e:#?}"), + ErrorCode::InternalError + ) + })?; + count += 1; + if count % 1000 == 0 { + pb.set_message(format!("{label}: {} blobs migrated...", count)); + } + } + + pb.finish_with_message(format!("{label}: {} blobs migrated", count)); + Ok(count) +} + +pub fn handle_migrate_v1(theme: &ColorfulTheme) { + println!( + "\n{}", + style("MIGRATION: Bichon v1.x Storage → v2.x (Fjall → bichon-blob)") + .bold() + .yellow() + ); + println!( + "{}\n", + style("This migrates blob storage from the fjall engine to bichon-blob.\n\ + Tantivy indexes and metadata (memdb) are NOT affected.") + .dim() + ); + + let root_dir: String = Input::with_theme(theme) + .with_prompt("Bichon root directory") + .with_initial_text("/var/lib/bichon") + .interact() + .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 fjall_path = data_base.join("bichon-storage"); + let blob_path = fjall_path.join("blobs"); + + if !fjall_path.exists() { + println!( + "{}", + style(format!( + "Fjall database not found at '{}'. Is this really a v1.x install?", + fjall_path.display() + )) + .red() + ); + return; + } + + if blob_path.exists() { + println!( + "{}", + style(format!( + "Target blob directory '{}' already exists.\n\ + If you have already migrated, you can remove the old fjall files manually.\n\ + Otherwise, delete this directory and re-run the migration.", + blob_path.display() + )) + .yellow() + ); + return; + } + + // Open old Fjall database (read-only by nature of the iter API) + println!("\n{}", style("Opening fjall database...").dim()); + let db = match Database::open(FjallConfig::new(&fjall_path)) { + Ok(db) => db, + Err(e) => { + println!( + "{}", + style(format!("Failed to open fjall database: {e:#?}")).red() + ); + return; + } + }; + + // Open new bichon-blob engine + println!("{}", style("Initializing bichon-blob engine...").dim()); + let mut config = Config::default(); + config.default_codec = Codec::Zstd; + config.compress_threshold = 1024; + config.flush_interval_secs = 0; + config.gc_interval_secs = 0; + + let engine = match Engine::open(&blob_path, config) { + Ok(e) => e, + Err(e) => { + println!( + "{}", + style(format!("Failed to open bichon-blob engine: {e:#?}")).red() + ); + return; + } + }; + + // Migrate email blobs + let email_count = match migrate_keyspace(&engine, &db, "email", "Email") { + Ok(n) => n, + Err(e) => { + println!("{}", style(format!("Email migration failed: {e:#?}")).red()); + let _ = engine.shutdown(); + return; + } + }; + + // Migrate attachment blobs + let attach_count = + match migrate_keyspace(&engine, &db, "attachments", "Attachment") { + Ok(n) => n, + Err(e) => { + println!( + "{}", + style(format!("Attachment migration failed: {e:#?}")).red() + ); + let _ = engine.shutdown(); + return; + } + }; + + // Flush and shutdown + println!("\n{}", style("Flushing and shutting down blob engine...").dim()); + if let Err(e) = engine.flush() { + println!("{}", style(format!("flush warning: {e:#?}")).yellow()); + } + if let Err(e) = engine.shutdown() { + println!( + "{}", + style(format!("shutdown error: {e:#?}")).red() + ); + return; + } + + // Write STORAGE_VERSION = 2 + if let Err(e) = write_storage_version(&root_dir, 2) { + println!( + "{}", + style(format!("Failed to write STORAGE_VERSION: {e:#?}")).red() + ); + return; + } + + println!( + "\n{}", + style(format!( + "Migration complete!\n Email blobs: {}\n Attachment blobs: {}\n Total: {}\n\n\ + The old fjall database at '{}' is no longer used.\n\ + You may delete it to free disk space after verifying everything works.", + email_count, + attach_count, + email_count + attach_count, + fjall_path.display() + )) + .green() + .bold() + ); +} + +#[cfg(test)] +mod tests { + use super::*; + use bichon_core::utils::compute_content_hash; + + // ── hex_key_to_raw ──────────────────────────────────────────────── + + #[test] + fn hex_key_to_raw_valid() { + // "hello" blake3 hex = 64 chars + let hash_hex = compute_content_hash(b"hello"); + assert_eq!(hash_hex.len(), 64); + + let raw = hex_key_to_raw(hash_hex.as_bytes()).unwrap(); + // Decoding 64 hex chars → 32 bytes + assert_eq!(raw.len(), 32); + // Round-trip: raw → hex should match original + assert_eq!(hex::encode(raw), hash_hex); + } + + #[test] + fn hex_key_to_raw_invalid_utf8() { + // 0xFF is not valid UTF-8 + let invalid = vec![0xFFu8; 64]; + let err = hex_key_to_raw(&invalid).unwrap_err(); + assert!(err.to_string().contains("invalid UTF-8")); + } + + #[test] + fn hex_key_to_raw_invalid_hex() { + // "zz" is valid UTF-8 but not valid hex + let invalid = b"zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz"; + let err = hex_key_to_raw(invalid).unwrap_err(); + assert!(err.to_string().contains("invalid hex")); + } + + #[test] + fn hex_key_to_raw_wrong_length() { + let short = b"abcd"; + let err = hex_key_to_raw(short).unwrap_err(); + assert!(err.to_string().contains("invalid hex")); + } + + #[test] + fn hex_key_to_raw_different_content() { + let a = hex_key_to_raw(compute_content_hash(b"a").as_bytes()).unwrap(); + let b = hex_key_to_raw(compute_content_hash(b"b").as_bytes()).unwrap(); + assert_ne!(a, b); + } + + // ── migrate_keyspace integration ────────────────────────────────── + + #[test] + fn migrate_keyspace_end_to_end() { + let tmp = tempfile::tempdir().unwrap(); + let fjall_path = tmp.path().join("fjall"); + let blob_path = tmp.path().join("blobs"); + + use fjall::KeyspaceCreateOptions; + + // --- Setup: create a Fjall database with test blobs --- + let fjall_db = Database::open(FjallConfig::new(&fjall_path)).unwrap(); + let ks = fjall_db + .keyspace("test_ks", KeyspaceCreateOptions::default) + .unwrap(); + + // Insert test blobs using hex string keys (matching v1.x convention) + let mut expected: Vec<(String, Vec)> = Vec::new(); + for i in 0..10 { + let data = format!("blob data {}", i).into_bytes(); + let hash = compute_content_hash(&data); + ks.insert(hash.as_bytes(), data.clone()).unwrap(); + expected.push((hash, data)); + } + + // --- Setup: create bichon-blob engine and run migration --- + { + let mut config = Config::default(); + config.flush_interval_secs = 0; + config.gc_interval_secs = 0; + let engine = Engine::open(&blob_path, config).unwrap(); + + let count = migrate_keyspace(&engine, &fjall_db, "test_ks", "Test").unwrap(); + assert_eq!(count, expected.len() as u64); + + engine.flush().unwrap(); + engine.shutdown().unwrap(); + // engine dropped here → LOCK released + } + + // --- Verify: re-open engine and check all blobs --- + let mut config = Config::default(); + config.flush_interval_secs = 0; + config.gc_interval_secs = 0; + let engine2 = Engine::open(&blob_path, config).unwrap(); + + for (hex_hash, expected_data) in &expected { + let mut raw_key = [0u8; 32]; + hex::decode_to_slice(hex_hash, &mut raw_key).unwrap(); + let got = engine2.get(&raw_key).unwrap(); + assert_eq!( + got.as_deref(), + Some(expected_data.as_slice()), + "mismatch for key {}", + hex_hash + ); + } + + // Verify non-existent key returns None + let fake_hash = compute_content_hash(b"nonexistent"); + let mut fake_key = [0u8; 32]; + hex::decode_to_slice(&fake_hash, &mut fake_key).unwrap(); + assert!(engine2.get(&fake_key).unwrap().is_none()); + + engine2.shutdown().unwrap(); + } + + #[test] + fn migrate_empty_keyspace() { + let tmp = tempfile::tempdir().unwrap(); + let fjall_path = tmp.path().join("fjall"); + let blob_path = tmp.path().join("blobs"); + + let fjall_db = Database::open(FjallConfig::new(&fjall_path)).unwrap(); + let _ks = fjall_db + .keyspace("empty_ks", fjall::KeyspaceCreateOptions::default) + .unwrap(); + + let mut config = Config::default(); + config.flush_interval_secs = 0; + config.gc_interval_secs = 0; + let engine = Engine::open(&blob_path, config).unwrap(); + + let count = migrate_keyspace(&engine, &fjall_db, "empty_ks", "Empty").unwrap(); + assert_eq!(count, 0); + + engine.shutdown().unwrap(); + } + + #[test] + fn hex_key_roundtrip_with_real_content_hash() { + // Simulate the exact data flow from v1.x to v2.x + let eml_data = b"From: sender@example.com\r\nSubject: Test\r\n\r\nHello world"; + let hash_hex = compute_content_hash(eml_data); // 64-char hex string + + // v1.x: key stored as hash_hex.as_bytes() + let fjall_key = hash_hex.as_bytes().to_vec(); + assert_eq!(fjall_key.len(), 64); + + // Migration: hex decode → raw 32 bytes + let raw_key = hex_key_to_raw(&fjall_key).unwrap(); + assert_eq!(raw_key.len(), 32); + + // v2.x: engine.put(raw_key, data) + // Verify round-trip: raw_key → hex → compare + let hex_roundtrip = hex::encode(raw_key); + assert_eq!(hex_roundtrip, hash_hex); + } +} diff --git a/crates/blob/src/meta.rs b/crates/blob/src/meta.rs index 77bc005..6bff72b 100644 --- a/crates/blob/src/meta.rs +++ b/crates/blob/src/meta.rs @@ -10,9 +10,10 @@ const META_VERSION: u32 = 2; // ── Helpers ──────────────────────────────────────────────────────────────── fn write_bin(path: &Path, value: &T) -> Result<()> { - let payload = bincode::serde::encode_to_vec(value, bincode::config::standard()).map_err(|e| { - crate::error::Error::CorruptMeta(format!("{}: bincode encode: {}", path.display(), e)) - })?; + let payload = + bincode::serde::encode_to_vec(value, bincode::config::standard()).map_err(|e| { + crate::error::Error::CorruptMeta(format!("{}: bincode encode: {}", path.display(), e)) + })?; let crc = checksum::crc32(&payload); let mut buf = Vec::with_capacity(8 + payload.len()); buf.extend_from_slice(&crc.to_le_bytes()); @@ -44,7 +45,7 @@ fn read_bin Deserialize<'de>>(path: &Path) -> Result { .map(|(v, _)| v) .map_err(|e| { crate::error::Error::CorruptMeta(format!("{}: bincode decode: {}", path.display(), e)) - }) + }) } // ── SegmentStats ─────────────────────────────────────────────────────────── @@ -108,20 +109,6 @@ impl GlobalMeta { if bin_path.exists() { return read_bin(&bin_path); } - // Migration from old JSON format - let json_path = store_root.join("meta.json"); - if json_path.exists() { - let data = std::fs::read_to_string(&json_path)?; - let meta: Self = serde_json::from_str(&data)?; - write_bin(&bin_path, &meta)?; - let _ = std::fs::remove_file(&json_path); - return Ok(meta); - } - // Migration from old global_meta.bin (v1, only had accounts list) - let old_path = store_root.join("global_meta.bin"); - if old_path.exists() { - let _ = std::fs::remove_file(&old_path); - } Ok(Self::new()) } diff --git a/crates/core/src/migrate/mod.rs b/crates/core/src/migrate/mod.rs index 10acd69..afbac36 100644 --- a/crates/core/src/migrate/mod.rs +++ b/crates/core/src/migrate/mod.rs @@ -48,13 +48,13 @@ pub fn is_tantivy_index_dir(dir: &PathBuf) -> std::io::Result { } /// Check whether the data layout is compatible with the current server. -/// Returns `false` only when legacy v0.3.7 data is detected and migration is required. +/// Returns `false` when legacy data (v0.3.7 or v1.x) is detected and migration is required. pub fn check_data_status() -> std::io::Result { let root_dir = PathBuf::from(&SETTINGS.bichon_root_dir); // 1. Version file takes precedence if let Some(version) = read_storage_version(&root_dir) { - return Ok(version >= 1); + return Ok(version >= CURRENT_STORAGE_VERSION); } // 2. No version file — check for existing v1.x-style storage (fjall era) @@ -66,9 +66,9 @@ pub fn check_data_status() -> std::io::Result { let new_storage_path = new_data_base.join("bichon-storage"); if is_dir_not_empty(&new_storage_path)? { - // Existing v1.x install predates version file — mark it + // Existing v1.x install predates version file — mark it as v1 let _ = write_storage_version(&root_dir, 1); - return Ok(true); + return Ok(false); // Needs migration: v1.x → v2.x } // 3. Check for legacy v0.3.7 Tantivy layout diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index f89e6de..4ba7e98 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -66,9 +66,11 @@ pub async fn run() -> BichonResult<()> { Ok(false) => { error!("Incompatible data format detected."); error!("Your data was created by an older version of Bichon and must be migrated before use."); - error!("Please stop the Bichon v0.3.7 service before migration."); error!("Please run: bichon-admin"); - error!("Documentation: https://github.com/rustmailer/bichon/wiki/Bichon-Data-Migration:-v0.3.7-%E2%86%92-v1.0"); + error!("Available migration options:"); + error!(" - Legacy v0.3.7 → v2.x (via v1.x)"); + error!(" - v1.x (Fjall) → v2.x (bichon-blob)"); + error!("Documentation: https://github.com/rustmailer/bichon/wiki"); return Err(raise_error!( "Legacy data layout detected".into(), ErrorCode::InternalError