diff --git a/Cargo.lock b/Cargo.lock index 715d9f0..05fcab5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -484,6 +484,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fd16c4719339c4530435d38e511904438d07cce7950afa3718a84ac36c10e89e" +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + [[package]] name = "ciborium" version = "0.2.2" @@ -799,6 +805,17 @@ dependencies = [ "syn 2.0.66", ] +[[package]] +name = "ctrlc" +version = "3.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "881c5d0a13b2f1498e2306e82cbada78390e152d4b1378fb28a84f4dcd0dc4f3" +dependencies = [ + "dispatch", + "nix 0.30.1", + "windows-sys 0.61.0", +] + [[package]] name = "darling" version = "0.20.9" @@ -2337,9 +2354,9 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646" [[package]] name = "libc" -version = "0.2.155" +version = "0.2.175" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97b3888a4aecf77e811145cadf6eef5901f4782c53886191b2f693f24761847c" +checksum = "6a82ae493e598baaea5209805c49bbf2ea7de956d50d7da0da1164f9c6d28543" [[package]] name = "libsqlite3-sys" @@ -2557,6 +2574,18 @@ dependencies = [ "memoffset", ] +[[package]] +name = "nix" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "74523f3a35e05aba87a1d978330aef40f67b0304ac79c1c00b294c9830543db6" +dependencies = [ + "bitflags 2.5.0", + "cfg-if", + "cfg_aliases 0.2.1", + "libc", +] + [[package]] name = "nodrop" version = "0.1.14" @@ -3077,6 +3106,7 @@ dependencies = [ name = "quiksearch" version = "0.1.0" dependencies = [ + "ctrlc", "dioxus", "dioxus-desktop", "dpc-pariter", @@ -3895,9 +3925,13 @@ checksum = "ba4f4a02a7a80d6f274636f0aa95c7e383b912d41fe721a31f29e29698585a4a" dependencies = [ "backtrace", "bytes", + "libc", + "mio", "num_cpus", "pin-project-lite", + "signal-hook-registry", "tokio-macros", + "windows-sys 0.48.0", ] [[package]] @@ -4451,6 +4485,12 @@ dependencies = [ "syn 2.0.66", ] +[[package]] +name = "windows-link" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "45e46c0661abb7180e7b9c281db115305d49ca1709ab8242adf09666d2173c65" + [[package]] name = "windows-result" version = "0.1.2" @@ -4487,6 +4527,15 @@ dependencies = [ "windows-targets 0.52.5", ] +[[package]] +name = "windows-sys" +version = "0.61.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e201184e40b2ede64bc2ea34968b28e33622acdbbf37104f0e4a33f7abe657aa" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-targets" version = "0.42.2" @@ -4703,7 +4752,7 @@ checksum = "8b717040ba9771fd88eb428c6ea6b555f8e734ff8534f02c13e8f10d97f5935e" dependencies = [ "base64", "block", - "cfg_aliases", + "cfg_aliases 0.1.1", "cocoa", "core-graphics", "crossbeam-channel", @@ -4824,7 +4873,7 @@ dependencies = [ "futures-sink", "futures-util", "hex", - "nix", + "nix 0.27.1", "ordered-stream", "rand 0.8.5", "serde", diff --git a/Cargo.toml b/Cargo.toml index 139860e..532a32c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,7 @@ tqdm = "0.7.0" walkdir = "2.5.0" zip = "0.6" quick-xml = "0.31" -tokio = { version = "1.0", features = ["time"] } +tokio = { version = "1.0", features = ["time", "signal"] } serde = { version = "1.0", features = ["derive"] } toml = "0.8" +ctrlc = "3.4" diff --git a/config_example.toml b/config_example.toml index 9e4c337..b8e0aeb 100644 --- a/config_example.toml +++ b/config_example.toml @@ -10,4 +10,7 @@ maximum_text_size = 524288 # Maximum file size to process for text extraction (bytes) maximum_file_size = 52428800 # Number of files to process in each batch -batch_size = 200 \ No newline at end of file +batch_size = 200 +# FTS5 tokenization method (e.g., 'trigram', 'porter', 'unicode61') +# Look here for more information https://www.sqlite.org/fts5.html#tokenizers +tokenize = "trigram" \ No newline at end of file diff --git a/src/config.rs b/src/config.rs index a7774cb..4cb9771 100644 --- a/src/config.rs +++ b/src/config.rs @@ -20,6 +20,7 @@ pub struct ProcessingConfig { pub maximum_text_size: usize, pub maximum_file_size: u64, pub batch_size: usize, + pub tokenize: String, } impl Default for Config { @@ -34,6 +35,7 @@ impl Default for Config { maximum_text_size: 1024 * 512, maximum_file_size: 1024 * 1024 * 50, batch_size: 200, + tokenize: "trigram".to_string(), }, } } diff --git a/src/file_handling.rs b/src/file_handling.rs index 9d93c19..dfdcde5 100644 --- a/src/file_handling.rs +++ b/src/file_handling.rs @@ -26,7 +26,7 @@ pub struct BatchUpdate { pub files_to_insert: Vec, } -const PLAINTEXT_EXTENSIONS_LIST: [&'static str; 86] = +pub const PLAINTEXT_EXTENSIONS_LIST: [&'static str; 86] = ["","txt","rtf","log", // Text Documents "csv", // Spreadsheet "sh","bat","cmd","bash","ps1","psm1","psd1","pssc","psrc", // Scripts @@ -44,7 +44,7 @@ const PLAINTEXT_EXTENSIONS_LIST: [&'static str; 86] = "wasm","wat", // Web Assembly ]; -const SUPPORTED_DOCUMENT_EXTENSIONS_LIST: [&'static str; 9] = +pub const SUPPORTED_DOCUMENT_EXTENSIONS_LIST: [&'static str; 9] = ["odt", "docx", "doc", // Office Documents "ppt", "pptx", "odp", // Presentation "xls", "xlsx", "ods"]; // Spreadsheet @@ -114,6 +114,21 @@ pub fn analyze_files_for_batch_update( } } +/// Safely truncate a string to at most max_bytes bytes while respecting UTF-8 character boundaries +fn safe_truncate_string(s: &str, max_bytes: usize) -> String { + if s.len() <= max_bytes { + return s.to_string(); + } + + // Find the last valid UTF-8 character boundary at or before max_bytes + let mut end = max_bytes; + while end > 0 && !s.is_char_boundary(end) { + end -= 1; + } + + s[..end].to_string() +} + /// Get a hash of a file by reading the first and last hash_length bytes of the file fn get_file_hash(size: u64, path: OsString, hash_length: usize) -> Result, std::io::Error> { let mut hasher = Sha256::new(); @@ -136,8 +151,8 @@ fn get_file_hash(size: u64, path: OsString, hash_length: usize) -> Result>, files_to_update: &[(DirEntry, FileMetadata)], stop_flag: &Arc>, @@ -176,91 +191,52 @@ pub fn process_batch_updates( let filename = entry.path().file_name() .and_then(|n| n.to_str()) .unwrap_or("unknown"); - callback(&format!("Updating file {}/{}: {}", global_index, total_files, filename)); + callback(&format!("Updating file metadata {}/{}: {}", global_index, total_files, filename)); } // Update progress counter if let Some(ref progress_cb) = progress_callback { progress_cb(global_index); } - let meta = entry.metadata().map_err(|e| format!("Failed to get metadata: {}", e))?; - if meta.is_dir() { - continue; - } - let fpath = match entry.path().canonicalize() { - Ok(fp) => fp.into_os_string(), - Err(_) => continue, - }; - - let fsize = meta.len(); - let fmodified = meta.modified() - .map_err(|e| format!("Failed to get modified time: {}", e))? - .duration_since(UNIX_EPOCH) - .map_err(|e| format!("Failed to calculate duration: {}", e))? - .as_secs(); - - let fhash = get_file_hash(fsize, fpath.clone(), config.processing.hash_length) - .map_err(|e| format!("Failed to calculate hash: {}", e))?; - - // Check stop flag after hash calculation - if *stop_flag.lock().unwrap() { - drop(tx); - drop(conn); - return Ok(()); - } - - // Update files table - tx.execute( - "UPDATE files SET size = ?1, moddate = ?2, hash = ?3 WHERE path = ?4", - params![fsize, fmodified, fhash, fpath.to_string_lossy()] - ).map_err(|e| format!("Failed to update file record: {}", e))?; - - // Delete old searchable text entry - tx.execute( - "DELETE FROM searchabletext WHERE path = ?1", - params![fpath.to_string_lossy()] - ).map_err(|e| format!("Failed to delete old searchable text: {}", e))?; - - // Insert new searchable text if applicable - if fsize <= config.processing.maximum_file_size { - let default_ext = OsString::new(); - let file_extension = entry.path().extension().unwrap_or(&default_ext) - .to_ascii_lowercase().to_str().unwrap_or("").to_string(); - let ext_str = file_extension.as_str(); - - if PLAINTEXT_EXTENSIONS_LIST.contains(&ext_str) { - if let Ok(file_string) = read_to_string(&fpath) { - let trimmed_file_string = if file_string.len() > config.processing.maximum_text_size { - file_string[..config.processing.maximum_text_size].to_string() - } else { - file_string - }; - - let fname = entry.path().file_name().unwrap().to_os_string(); - tx.execute( - "INSERT INTO searchabletext VALUES (?1, ?2, ?3)", - params![fname.to_str(), fpath.to_string_lossy(), trimmed_file_string] - ).map_err(|e| format!("Failed to insert searchable text: {}", e))?; - } - } else if SUPPORTED_DOCUMENT_EXTENSIONS_LIST.contains(&ext_str) { - if let Ok(extracted_text) = extract_document_text(&fpath, ext_str) { - if !extracted_text.trim().is_empty() { - let trimmed_file_string = if extracted_text.len() > config.processing.maximum_text_size { - extracted_text[..config.processing.maximum_text_size].to_string() - } else { - extracted_text - }; - - let fname = entry.path().file_name().unwrap().to_os_string(); - tx.execute( - "INSERT INTO searchabletext VALUES (?1, ?2, ?3)", - params![fname.to_str(), fpath.to_string_lossy(), trimmed_file_string] - ).map_err(|e| format!("Failed to insert document text: {}", e))?; - } - } + let meta = entry.metadata().map_err(|e| format!("Failed to get metadata: {}", e))?; + if meta.is_dir() { + continue; } - } + + let fpath = match entry.path().canonicalize() { + Ok(fp) => fp.into_os_string(), + Err(_) => continue, + }; + + let fsize = meta.len(); + let fmodified = meta.modified() + .map_err(|e| format!("Failed to get modified time: {}", e))? + .duration_since(UNIX_EPOCH) + .map_err(|e| format!("Failed to calculate duration: {}", e))? + .as_secs(); + + let fhash = get_file_hash(fsize, fpath.clone(), config.processing.hash_length) + .map_err(|e| format!("Failed to calculate hash: {}", e))?; + + // Check stop flag after hash calculation + if *stop_flag.lock().unwrap() { + drop(tx); + drop(conn); + return Ok(()); + } + + // Update files table + tx.execute( + "UPDATE files SET size = ?1, moddate = ?2, hash = ?3 WHERE path = ?4", + params![fsize, fmodified, fhash, fpath.to_string_lossy()] + ).map_err(|e| format!("Failed to update file record: {}", e))?; + + // Delete old searchable text entry for this file (will be re-added in text indexing phase) + tx.execute( + "DELETE FROM searchabletext WHERE path = ?1", + params![fpath.to_string_lossy()] + ).map_err(|e| format!("Failed to delete old searchable text: {}", e))?; } tx.commit().map_err(|e| format!("Failed to commit transaction: {}", e))?; @@ -269,8 +245,8 @@ pub fn process_batch_updates( Ok(()) } -/// Process new files in batch with transaction -pub fn process_batch_inserts( +/// Process new files in batch with transaction - files table only (no text extraction) +pub fn process_batch_inserts_files_only( conn_mutex: &Arc>, files_to_insert: &[DirEntry], stop_flag: &Arc>, @@ -306,80 +282,162 @@ pub fn process_batch_inserts( // Update status with current file if let Some(ref callback) = status_callback { - let file_path = entry.path().to_str() + let filename = entry.path().file_name() + .and_then(|n| n.to_str()) .unwrap_or("unknown"); - callback(&format!("Indexing file: {}", file_path)); + callback(&format!("Indexing file metadata {}/{}: {}", global_index, total_files, filename)); } // Update progress counter if let Some(ref progress_cb) = progress_callback { progress_cb(global_index); } - let meta = entry.metadata().map_err(|e| format!("Failed to get metadata: {}", e))?; - if meta.is_dir() { - continue; - } - let fpath = match entry.path().canonicalize() { - Ok(fp) => fp.into_os_string(), - Err(_) => continue, - }; - - let fsize = meta.len(); - let fmodified = meta.modified() - .map_err(|e| format!("Failed to get modified time: {}", e))? - .duration_since(UNIX_EPOCH) - .map_err(|e| format!("Failed to calculate duration: {}", e))? - .as_secs(); - - let fhash = get_file_hash(fsize, fpath.clone(), config.processing.hash_length) - .map_err(|e| format!("Failed to calculate hash: {}", e))?; - - let fname = entry.path().file_name().unwrap().to_os_string(); - - // Insert into files table - tx.execute( - "INSERT INTO files VALUES (?1, ?2, ?3, ?4, ?5)", - params![fname.to_str(), fpath.to_string_lossy(), fsize, fmodified, fhash] - ).map_err(|e| format!("Failed to insert file record: {}", e))?; - - // Insert searchable text if applicable - if fsize <= config.processing.maximum_file_size { - let default_ext = OsString::new(); - let file_extension = entry.path().extension().unwrap_or(&default_ext) - .to_ascii_lowercase().to_str().unwrap_or("").to_string(); - let ext_str = file_extension.as_str(); - - if PLAINTEXT_EXTENSIONS_LIST.contains(&ext_str) { - if let Ok(file_string) = read_to_string(&fpath) { - let trimmed_file_string = if file_string.len() > config.processing.maximum_text_size { - file_string[..config.processing.maximum_text_size].to_string() - } else { - file_string - }; - - tx.execute( - "INSERT INTO searchabletext VALUES (?1, ?2, ?3)", - params![fname.to_str(), fpath.to_string_lossy(), trimmed_file_string] - ).map_err(|e| format!("Failed to insert searchable text: {}", e))?; - } - } else if SUPPORTED_DOCUMENT_EXTENSIONS_LIST.contains(&ext_str) { - if let Ok(extracted_text) = extract_document_text(&fpath, ext_str) { - if !extracted_text.trim().is_empty() { - let trimmed_file_string = if extracted_text.len() > config.processing.maximum_text_size { - extracted_text[..config.processing.maximum_text_size].to_string() - } else { - extracted_text - }; - - tx.execute( - "INSERT INTO searchabletext VALUES (?1, ?2, ?3)", - params![fname.to_str(), fpath.to_string_lossy(), trimmed_file_string] - ).map_err(|e| format!("Failed to insert document text: {}", e))?; - } - } + let meta = entry.metadata().map_err(|e| format!("Failed to get metadata: {}", e))?; + if meta.is_dir() { + continue; + } + + let fpath = match entry.path().canonicalize() { + Ok(fp) => fp.into_os_string(), + Err(_) => continue, + }; + + let fsize = meta.len(); + let fmodified = meta.modified() + .map_err(|e| format!("Failed to get modified time: {}", e))? + .duration_since(UNIX_EPOCH) + .map_err(|e| format!("Failed to calculate duration: {}", e))? + .as_secs(); + + let fhash = get_file_hash(fsize, fpath.clone(), config.processing.hash_length) + .map_err(|e| format!("Failed to calculate hash: {}", e))?; + + let fname = entry.path().file_name().unwrap().to_os_string(); + + // Insert into files table + tx.execute( + "INSERT INTO files VALUES (?1, ?2, ?3, ?4, ?5)", + params![fname.to_str(), fpath.to_string_lossy(), fsize, fmodified, fhash] + ).map_err(|e| format!("Failed to insert file record: {}", e))?; + } + + tx.commit().map_err(|e| format!("Failed to commit transaction: {}", e))?; + } + + Ok(()) +} + +/// Process text indexing for files - adds entries to searchabletext table +pub fn process_text_indexing( + conn_mutex: &Arc>, + stop_flag: &Arc>, + status_callback: Option>, + progress_callback: Option>, + config: &Config +) -> Result<(), String> { + let conn = conn_mutex.lock().unwrap(); + + // Get all files from the files table that don't have corresponding searchabletext entries + let mut stmt = conn.prepare( + "SELECT f.name, f.path, f.size FROM files f + LEFT JOIN searchabletext s ON f.path = s.path + WHERE s.path IS NULL AND f.size <= ?1" + ).map_err(|e| format!("Failed to prepare statement: {}", e))?; + + let file_rows = stmt.query_map([config.processing.maximum_file_size], |row| { + Ok(( + row.get::<_, String>(0)?, // name + row.get::<_, String>(1)?, // path + row.get::<_, u64>(2)? // size + )) + }).map_err(|e| format!("Failed to query files: {}", e))?; + + let files_to_process: Vec<_> = file_rows.collect::, _>>() + .map_err(|e| format!("Failed to collect files: {}", e))?; + + drop(stmt); + drop(conn); + + let total_files = files_to_process.len(); + let batch_size = config.processing.batch_size; + + // Process files in batches + for (batch_idx, batch) in files_to_process.chunks(batch_size).enumerate() { + // Check stop flag at the start of each batch + if *stop_flag.lock().unwrap() { + return Ok(()); + } + + let conn = conn_mutex.lock().unwrap(); + let tx = conn.unchecked_transaction().map_err(|e| format!("Failed to begin transaction: {}", e))?; + + for (i, (fname, fpath, _fsize)) in batch.iter().enumerate() { + let global_index = batch_idx * batch_size + i + 1; + + // Check stop flag + if *stop_flag.lock().unwrap() { + drop(tx); + drop(conn); + return Ok(()); + } + + // Update status + if let Some(ref callback) = status_callback { + callback(&format!("Extracting text for search indexing {}/{}: {}", global_index, total_files, fname)); + } + + // Update progress counter + if let Some(ref progress_cb) = progress_callback { + progress_cb(global_index); + } + + let path = std::path::Path::new(fpath); + let default_ext = OsString::new(); + let file_extension = path.extension().unwrap_or(&default_ext) + .to_ascii_lowercase().to_str().unwrap_or("").to_string(); + let ext_str = file_extension.as_str(); + + // Process text content with error handling for individual files + let text_result = if PLAINTEXT_EXTENSIONS_LIST.contains(&ext_str) { + match read_to_string(fpath) { + Ok(file_string) => { + let trimmed_file_string = safe_truncate_string(&file_string, config.processing.maximum_text_size); + Some(trimmed_file_string) + } + Err(e) => { + // eprintln!("Warning: Failed to read plaintext file {}: {}", fpath, e); + None + } + } + } else if SUPPORTED_DOCUMENT_EXTENSIONS_LIST.contains(&ext_str) { + match extract_document_text(&std::ffi::OsString::from(fpath), ext_str) { + Ok(extracted_text) => { + if !extracted_text.trim().is_empty() { + let trimmed_file_string = safe_truncate_string(&extracted_text, config.processing.maximum_text_size); + Some(trimmed_file_string) + } else { + None + } + } + Err(e) => { + eprintln!("Warning: Failed to extract text from document {}: {}", fpath, e); + None + } + } + } else { + None + }; + + // Insert the text content if we successfully extracted it + if let Some(text_content) = text_result { + if let Err(e) = tx.execute( + "INSERT INTO searchabletext VALUES (?1, ?2, ?3)", + params![fname, fpath, text_content] + ) { + eprintln!("Warning: Failed to insert searchable text for {}: {}", fpath, e); + } } - } } tx.commit().map_err(|e| format!("Failed to commit transaction: {}", e))?; diff --git a/src/frontend.rs b/src/frontend.rs index dd423ee..8cad342 100644 --- a/src/frontend.rs +++ b/src/frontend.rs @@ -105,7 +105,7 @@ pub fn App(props: AppProps) -> Element { speed_tracker_clone.set(SpeedTracker::new()); "Idle".to_string() }, - IndexingStatus::Running { files_processed, total_files, current_file, start_time } => { + IndexingStatus::RunningFileIndex { files_processed, total_files, current_file, start_time } => { // Add data point to speed tracker speed_tracker_clone.with_mut(|tracker| { tracker.add_data_point(files_processed); @@ -129,7 +129,7 @@ pub fn App(props: AppProps) -> Element { (files_processed as f64 / total as f64 * 100.0) as u32 } else { 0 }; format!( - "Running: {}/{} files ({}%) - {:.1}s elapsed{}\n{}", + "Phase 1 - File Index: {}/{} files ({}%) - {:.1}s elapsed{}\n{}", files_processed, total, percentage, @@ -139,7 +139,49 @@ pub fn App(props: AppProps) -> Element { ) } else { format!( - "Running: {} files processed - {:.1}s elapsed{}\n{}", + "Phase 1 - File Index: {} files processed - {:.1}s elapsed{}\n{}", + files_processed, + elapsed.as_secs_f64(), + speed_display, + current_file_display + ) + } + } + IndexingStatus::RunningTextIndex { files_processed, total_files, current_file, start_time } => { + // Add data point to speed tracker + speed_tracker_clone.with_mut(|tracker| { + tracker.add_data_point(files_processed); + }); + + let elapsed = start_time.elapsed(); + let current_file_display = current_file + .as_ref() + .map(|f| format!("Current: {}", f)) + .unwrap_or_default(); + + // Calculate speed + let speed_display = speed_tracker_clone.with(|tracker| { + tracker.calculate_files_per_second() + .map(|fps| format!(" - {:.1} files/sec", fps)) + .unwrap_or_default() + }); + + if let Some(total) = total_files { + let percentage = if total > 0 { + (files_processed as f64 / total as f64 * 100.0) as u32 + } else { 0 }; + format!( + "Phase 2 - Text Index: {}/{} files ({}%) - {:.1}s elapsed{}\n{}", + files_processed, + total, + percentage, + elapsed.as_secs_f64(), + speed_display, + current_file_display + ) + } else { + format!( + "Phase 2 - Text Index: {} files processed - {:.1}s elapsed{}\n{}", files_processed, elapsed.as_secs_f64(), speed_display, diff --git a/src/indexing.rs b/src/indexing.rs index 47feb7d..ca59c25 100644 --- a/src/indexing.rs +++ b/src/indexing.rs @@ -4,13 +4,19 @@ use std::time::Instant; use walkdir::WalkDir; use rusqlite::{Connection, params}; -use crate::file_handling::{load_existing_files, analyze_files_for_batch_update, process_batch_updates, process_batch_inserts}; +use crate::file_handling::{load_existing_files, analyze_files_for_batch_update, process_batch_updates_files_only, process_batch_inserts_files_only, process_text_indexing}; use crate::config::Config; #[derive(Debug, Clone)] pub enum IndexingStatus { Idle, - Running { + RunningFileIndex { + files_processed: usize, + total_files: Option, + current_file: Option, + start_time: Instant, + }, + RunningTextIndex { files_processed: usize, total_files: Option, current_file: Option, @@ -30,9 +36,11 @@ pub enum IndexingCommand { Stop, } +#[derive(Debug)] pub struct IndexingService { status: Arc>, command_tx: mpsc::Sender, + db_connection: Arc>>>>, _handle: thread::JoinHandle<()>, } @@ -68,15 +76,18 @@ impl IndexingService { pub fn new() -> Self { let status = Arc::new(Mutex::new(IndexingStatus::Idle)); let (command_tx, command_rx) = mpsc::channel(); + let db_connection = Arc::new(Mutex::new(None)); let status_clone = status.clone(); + let db_connection_clone = db_connection.clone(); let handle = thread::spawn(move || { - Self::indexing_thread(status_clone, command_rx); + Self::indexing_thread(status_clone, command_rx, db_connection_clone); }); IndexingService { status, command_tx, + db_connection, _handle: handle, } } @@ -88,15 +99,56 @@ impl IndexingService { } pub fn stop_indexing(&self) -> Result<(), String> { + // First send the stop command self.command_tx .send(IndexingCommand::Stop) - .map_err(|e| format!("Failed to send stop command: {}", e)) + .map_err(|e| format!("Failed to send stop command: {}", e))?; + + // Wait for indexing to transition to stopping state + let mut attempts = 0; + while attempts < 50 { // Wait up to 5 seconds + match self.get_status() { + IndexingStatus::Stopping => break, + IndexingStatus::Idle => return Ok(()), // Already stopped + IndexingStatus::Error(_) => return Ok(()), // Consider error state as stopped + _ => { + std::thread::sleep(std::time::Duration::from_millis(100)); + attempts += 1; + } + } + } + + // Flush and close database connection if it exists + if let Ok(mut db_opt) = self.db_connection.lock() { + if let Some(db_conn_arc) = db_opt.take() { + if let Ok(conn) = db_conn_arc.lock() { + // Re-enable journal mode and synchronous writes for proper flushing + let _ = conn.execute_batch( + "PRAGMA journal_mode = DELETE; + PRAGMA synchronous = FULL;" + ); + + // Force a checkpoint to flush any remaining WAL data + let _ = conn.execute("PRAGMA wal_checkpoint(FULL);", ()); + + // Explicitly close the connection by dropping it + drop(conn); + } + } + } + + Ok(()) } pub fn get_status(&self) -> IndexingStatus { self.status.lock().unwrap().clone() } + /// Force graceful shutdown - used for signal handling + pub fn graceful_shutdown(&self) -> Result<(), String> { + self.stop_indexing() + } + /// Check if configuration changes require index recreation pub fn check_config_validation(&self, db_path: &str, config: &Config, indexing_path: &str) -> Result>, String> { let conn = Connection::open(db_path) @@ -124,7 +176,7 @@ impl IndexingService { while attempts < 50 { // Wait up to 5 seconds match self.get_status() { IndexingStatus::Idle => break, - IndexingStatus::Stopping | IndexingStatus::Running { .. } => { + IndexingStatus::Stopping | IndexingStatus::RunningFileIndex { .. } | IndexingStatus::RunningTextIndex { .. } => { std::thread::sleep(std::time::Duration::from_millis(100)); attempts += 1; } @@ -141,14 +193,18 @@ impl IndexingService { Ok(()) } - fn indexing_thread(status: Arc>, command_rx: mpsc::Receiver) { + fn indexing_thread( + status: Arc>, + command_rx: mpsc::Receiver, + db_connection: Arc>>>> + ) { let stop_flag = Arc::new(Mutex::new(false)); let mut indexing_handle: Option> = None; while let Ok(command) = command_rx.recv() { match command { IndexingCommand::Start { path, db_path, config } => { - if matches!(*status.lock().unwrap(), IndexingStatus::Running { .. }) { + if matches!(*status.lock().unwrap(), IndexingStatus::RunningFileIndex { .. } | IndexingStatus::RunningTextIndex { .. }) { continue; // Already running } @@ -158,7 +214,7 @@ impl IndexingService { } *stop_flag.lock().unwrap() = false; - *status.lock().unwrap() = IndexingStatus::Running { + *status.lock().unwrap() = IndexingStatus::RunningFileIndex { files_processed: 0, total_files: None, current_file: None, @@ -172,8 +228,9 @@ impl IndexingService { let db_path_owned = db_path.clone(); let config_owned = config.clone(); + let db_connection_clone = db_connection.clone(); indexing_handle = Some(thread::spawn(move || { - if let Err(e) = Self::run_indexing(&status_clone, &path_owned, &db_path_owned, &stop_flag_clone, &config_owned) { + if let Err(e) = Self::run_indexing(&status_clone, &path_owned, &db_path_owned, &stop_flag_clone, &config_owned, &db_connection_clone) { *status_clone.lock().unwrap() = IndexingStatus::Error(e); } else { // Only set to Idle if we weren't stopped @@ -181,10 +238,15 @@ impl IndexingService { *status_clone.lock().unwrap() = IndexingStatus::Idle; } } + + // Clear the database connection when indexing completes + if let Ok(mut db_opt) = db_connection_clone.lock() { + *db_opt = None; + } })); } IndexingCommand::Stop => { - if matches!(*status.lock().unwrap(), IndexingStatus::Running { .. }) { + if matches!(*status.lock().unwrap(), IndexingStatus::RunningFileIndex { .. } | IndexingStatus::RunningTextIndex { .. }) { *status.lock().unwrap() = IndexingStatus::Stopping; *stop_flag.lock().unwrap() = true; } @@ -204,6 +266,7 @@ impl IndexingService { db_path: &str, stop_flag: &Arc>, config: &Config, + db_connection: &Arc>>>>, ) -> Result<(), String> { // Set up database let conn = Connection::open(db_path) @@ -228,11 +291,12 @@ impl IndexingService { ) .map_err(|e| format!("Failed to create files table: {}", e))?; - conn.execute( - "CREATE VIRTUAL TABLE IF NOT EXISTS searchabletext USING fts5 (name, path, text, tokenize = 'trigram');", - (), - ) - .map_err(|e| format!("Failed to create searchabletext table: {}", e))?; + let create_fts_sql = format!( + "CREATE VIRTUAL TABLE IF NOT EXISTS searchabletext USING fts5 (name, path, text, tokenize = '{}');", + config.processing.tokenize + ); + conn.execute(&create_fts_sql, ()) + .map_err(|e| format!("Failed to create searchabletext table: {}", e))?; conn.execute( "CREATE TABLE IF NOT EXISTS config_validation ( @@ -253,6 +317,11 @@ impl IndexingService { }; let conn_mutex = Arc::new(Mutex::new(conn)); + + // Store the database connection for proper cleanup on stop + if let Ok(mut db_opt) = db_connection.lock() { + *db_opt = Some(conn_mutex.clone()); + } // Collect all file entries let walker = WalkDir::new(path).into_iter(); @@ -265,7 +334,7 @@ impl IndexingService { // Update status with total file count if let Ok(mut status_guard) = status.lock() { - if let IndexingStatus::Running { ref mut total_files, .. } = *status_guard { + if let IndexingStatus::RunningFileIndex { ref mut total_files, .. } = *status_guard { *total_files = Some(total_file_count); } } @@ -277,7 +346,7 @@ impl IndexingService { // Update status to show actual work needed if let Ok(mut status_guard) = status.lock() { - if let IndexingStatus::Running { ref mut total_files, .. } = *status_guard { + if let IndexingStatus::RunningFileIndex { ref mut total_files, .. } = *status_guard { *total_files = Some(total_work); } } @@ -287,7 +356,7 @@ impl IndexingService { // Process updated files in batches if !batch_update.files_to_update.is_empty() { if let Ok(mut status_guard) = status.lock() { - if let IndexingStatus::Running { ref mut current_file, .. } = *status_guard { + if let IndexingStatus::RunningFileIndex { ref mut current_file, .. } = *status_guard { *current_file = Some(format!("Updating {} modified files...", batch_update.files_to_update.len())); } } @@ -296,7 +365,7 @@ impl IndexingService { let status_clone_1 = status.clone(); let status_callback = Box::new(move |file_status: &str| { if let Ok(mut status_guard) = status_clone_1.lock() { - if let IndexingStatus::Running { ref mut current_file, .. } = *status_guard { + if let IndexingStatus::RunningFileIndex { ref mut current_file, .. } = *status_guard { *current_file = Some(file_status.to_string()); } } @@ -307,20 +376,20 @@ impl IndexingService { let base_work_completed = work_completed; let progress_callback = Box::new(move |current_index: usize| { if let Ok(mut status_guard) = status_clone_2.lock() { - if let IndexingStatus::Running { ref mut files_processed, .. } = *status_guard { + if let IndexingStatus::RunningFileIndex { ref mut files_processed, .. } = *status_guard { *files_processed = base_work_completed + current_index; } } }); - if let Err(e) = process_batch_updates(&conn_mutex, &batch_update.files_to_update, &stop_flag, Some(status_callback), Some(progress_callback), config) { + if let Err(e) = process_batch_updates_files_only(&conn_mutex, &batch_update.files_to_update, &stop_flag, Some(status_callback), Some(progress_callback), config) { return Err(format!("Failed to process batch updates: {}", e)); } work_completed += batch_update.files_to_update.len(); if let Ok(mut status_guard) = status.lock() { - if let IndexingStatus::Running { ref mut files_processed, .. } = *status_guard { + if let IndexingStatus::RunningFileIndex { ref mut files_processed, .. } = *status_guard { *files_processed = work_completed; } } @@ -337,7 +406,7 @@ impl IndexingService { // Process new files in batches if !batch_update.files_to_insert.is_empty() { if let Ok(mut status_guard) = status.lock() { - if let IndexingStatus::Running { ref mut current_file, .. } = *status_guard { + if let IndexingStatus::RunningFileIndex { ref mut current_file, .. } = *status_guard { *current_file = Some(format!("Indexing {} new files...", batch_update.files_to_insert.len())); } } @@ -346,7 +415,7 @@ impl IndexingService { let status_clone_3 = status.clone(); let status_callback = Box::new(move |file_status: &str| { if let Ok(mut status_guard) = status_clone_3.lock() { - if let IndexingStatus::Running { ref mut current_file, .. } = *status_guard { + if let IndexingStatus::RunningFileIndex { ref mut current_file, .. } = *status_guard { *current_file = Some(file_status.to_string()); } } @@ -357,35 +426,85 @@ impl IndexingService { let base_work_completed = work_completed; let progress_callback = Box::new(move |current_index: usize| { if let Ok(mut status_guard) = status_clone_4.lock() { - if let IndexingStatus::Running { ref mut files_processed, .. } = *status_guard { + if let IndexingStatus::RunningFileIndex { ref mut files_processed, .. } = *status_guard { *files_processed = base_work_completed + current_index; } } }); - if let Err(e) = process_batch_inserts(&conn_mutex, &batch_update.files_to_insert, &stop_flag, Some(status_callback), Some(progress_callback), config) { + if let Err(e) = process_batch_inserts_files_only(&conn_mutex, &batch_update.files_to_insert, &stop_flag, Some(status_callback), Some(progress_callback), config) { return Err(format!("Failed to process batch inserts: {}", e)); } work_completed += batch_update.files_to_insert.len(); if let Ok(mut status_guard) = status.lock() { - if let IndexingStatus::Running { ref mut files_processed, .. } = *status_guard { + if let IndexingStatus::RunningFileIndex { ref mut files_processed, .. } = *status_guard { *files_processed = work_completed; } } } - // If no incremental work was needed, show completion status + // If no incremental work was needed, show completion status for file indexing phase if total_work == 0 { if let Ok(mut status_guard) = status.lock() { - if let IndexingStatus::Running { ref mut current_file, ref mut files_processed, .. } = *status_guard { - *current_file = Some("Index is up to date".to_string()); + if let IndexingStatus::RunningFileIndex { ref mut current_file, ref mut files_processed, .. } = *status_guard { + *current_file = Some("File index is up to date".to_string()); *files_processed = total_file_count; } } } + // Check for stop signal before starting text indexing + if *stop_flag.lock().unwrap() { + if let Ok(mut status_guard) = status.lock() { + *status_guard = IndexingStatus::Idle; + } + return Ok(()); + } + + // Phase 2: Text indexing + if let Ok(mut status_guard) = status.lock() { + *status_guard = IndexingStatus::RunningTextIndex { + files_processed: 0, + total_files: None, + current_file: Some("Starting text indexing...".to_string()), + start_time: Instant::now(), + }; + } + + // Create status callback for text indexing + let status_clone_5 = status.clone(); + let text_status_callback = Box::new(move |file_status: &str| { + if let Ok(mut status_guard) = status_clone_5.lock() { + if let IndexingStatus::RunningTextIndex { ref mut current_file, .. } = *status_guard { + *current_file = Some(file_status.to_string()); + } + } + }); + + // Create progress callback for text indexing + let status_clone_6 = status.clone(); + let text_progress_callback = Box::new(move |current_index: usize| { + if let Ok(mut status_guard) = status_clone_6.lock() { + if let IndexingStatus::RunningTextIndex { ref mut files_processed, .. } = *status_guard { + *files_processed = current_index; + } + } + }); + + // Process text indexing + if let Err(e) = process_text_indexing(&conn_mutex, &stop_flag, Some(text_status_callback), Some(text_progress_callback), config) { + return Err(format!("Failed to process text indexing: {}", e)); + } + + // Mark text indexing as complete + if let Ok(mut status_guard) = status.lock() { + if let IndexingStatus::RunningTextIndex { ref mut current_file, .. } = *status_guard { + *current_file = Some("Text indexing complete".to_string()); + } + } + Ok(()) } @@ -393,9 +512,11 @@ impl IndexingService { /// Critical configuration changes that require index recreation: /// - hash_length: affects file hash computation, invalidates existing file metadata /// - indexing_path: changes the scope of indexed files + /// - tokenize: changes FTS5 tokenization, invalidates text search index fn validate_config(conn: &Connection, config: &Config, indexing_path: &str) -> Result>, String> { // Critical configuration values that require index recreation let hash_length = config.processing.hash_length.to_string(); + let tokenize = config.processing.tokenize.clone(); let normalized_path = std::path::Path::new(indexing_path) .canonicalize() .unwrap_or_else(|_| std::path::PathBuf::from(indexing_path)) @@ -405,8 +526,9 @@ impl IndexingService { // Check stored configuration values let mut stored_hash_length: Option = None; let mut stored_indexing_path: Option = None; + let mut stored_tokenize: Option = None; - if let Ok(mut stmt) = conn.prepare("SELECT key, value FROM config_validation WHERE key IN ('hash_length', 'indexing_path')") { + if let Ok(mut stmt) = conn.prepare("SELECT key, value FROM config_validation WHERE key IN ('hash_length', 'indexing_path', 'tokenize')") { if let Ok(rows) = stmt.query_map([], |row| { let key: String = row.get(0)?; let value: String = row.get(1)?; @@ -416,6 +538,7 @@ impl IndexingService { match row.0.as_str() { "hash_length" => stored_hash_length = Some(row.1), "indexing_path" => stored_indexing_path = Some(row.1), + "tokenize" => stored_tokenize = Some(row.1), _ => {} } } @@ -425,8 +548,9 @@ impl IndexingService { // Check if configuration is invalid let hash_length_changed = stored_hash_length.as_ref().map_or(false, |stored| stored != &hash_length); let indexing_path_changed = stored_indexing_path.as_ref().map_or(false, |stored| stored != &normalized_path); + let tokenize_changed = stored_tokenize.as_ref().map_or(false, |stored| stored != &tokenize); - if hash_length_changed || indexing_path_changed { + if hash_length_changed || indexing_path_changed || tokenize_changed { let mut changes = Vec::new(); if hash_length_changed { changes.push(format!("hash_length: {} -> {}", @@ -436,6 +560,10 @@ impl IndexingService { changes.push(format!("indexing_path: {} -> {}", stored_indexing_path.unwrap_or_else(|| "unknown".to_string()), normalized_path)); } + if tokenize_changed { + changes.push(format!("tokenize: {} -> {}", + stored_tokenize.unwrap_or_else(|| "unknown".to_string()), tokenize)); + } return Ok(Some(changes)); } @@ -448,6 +576,7 @@ impl IndexingService { /// Updates stored configuration values without clearing the index fn update_config(conn: &Connection, config: &Config, indexing_path: &str) -> Result<(), String> { let hash_length = config.processing.hash_length.to_string(); + let tokenize = config.processing.tokenize.clone(); let normalized_path = std::path::Path::new(indexing_path) .canonicalize() .unwrap_or_else(|_| std::path::PathBuf::from(indexing_path)) @@ -464,11 +593,23 @@ impl IndexingService { "INSERT OR REPLACE INTO config_validation (key, value) VALUES ('indexing_path', ?1)", params![normalized_path], ).map_err(|e| format!("Failed to store indexing_path config: {}", e))?; + + conn.execute( + "INSERT OR REPLACE INTO config_validation (key, value) VALUES ('tokenize', ?1)", + params![tokenize], + ).map_err(|e| format!("Failed to store tokenize config: {}", e))?; Ok(()) } } +impl Drop for IndexingService { + fn drop(&mut self) { + // Ensure graceful shutdown when the service is dropped + let _ = self.stop_indexing(); + } +} + impl Default for IndexingService { fn default() -> Self { Self::new() diff --git a/src/main.rs b/src/main.rs index d9cda52..f2bca0a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use dioxus::prelude::*; mod frontend; @@ -7,7 +7,25 @@ mod document_extraction; mod indexing; mod config; +// Global indexing service for signal handling +static INDEXING_SERVICE: OnceLock> = OnceLock::new(); + fn main() { + // Initialize global indexing service + let indexing_service = Arc::new(indexing::IndexingService::new()); + INDEXING_SERVICE.set(indexing_service.clone()).expect("Failed to set global indexing service"); + + // Set up Ctrl-C signal handler + ctrlc::set_handler(|| { + eprintln!("Received Ctrl-C, shutting down gracefully..."); + if let Some(service) = INDEXING_SERVICE.get() { + if let Err(e) = service.graceful_shutdown() { + eprintln!("Error during graceful shutdown: {}", e); + } + } + std::process::exit(0); + }).expect("Error setting Ctrl-C handler"); + launch(app); } @@ -20,7 +38,7 @@ fn app() -> Element { } }; - let indexing_service = Arc::new(indexing::IndexingService::new()); + let indexing_service = INDEXING_SERVICE.get().expect("Indexing service not initialized").clone(); rsx! { frontend::App {