diff --git a/src/file_handling.rs b/src/file_handling.rs index cf18fab..ca9edac 100644 --- a/src/file_handling.rs +++ b/src/file_handling.rs @@ -3,6 +3,7 @@ use std::ffi::OsString; use std::fs::{File,read_to_string}; use std::io::{Read, Seek, SeekFrom}; use std::time::UNIX_EPOCH; +use std::collections::HashMap; use sha2::{Sha256, Digest}; use walkdir::DirEntry; @@ -10,6 +11,19 @@ use rusqlite::{params, Connection}; use crate::document_extraction::extract_document_text; +#[derive(Debug, Clone)] +pub struct FileMetadata { + pub path: String, + pub size: u64, + pub moddate: u64, + pub hash: Vec, +} + +#[derive(Debug)] +pub struct BatchUpdate { + pub files_to_update: Vec<(DirEntry, FileMetadata)>, + pub files_to_insert: Vec, +} const HASHLEN:usize = 1024 * 8; const MAXIMUM_TEXT_SIZE:usize = 1024 * 512; @@ -37,6 +51,71 @@ const SUPPORTED_DOCUMENT_EXTENSIONS_LIST: [&'static str; 9] = "ppt", "pptx", "odp", // Presentation "xls", "xlsx", "ods"]; // Spreadsheet +/// Load existing file metadata from database indexed by path +pub fn load_existing_files(conn: &Connection) -> Result, rusqlite::Error> { + let mut existing_files = HashMap::new(); + let mut stmt = conn.prepare("SELECT path, size, moddate, hash FROM files")?; + let rows = stmt.query_map([], |row| { + Ok(FileMetadata { + path: row.get(0)?, + size: row.get(1)?, + moddate: row.get(2)?, + hash: row.get(3)?, + }) + })?; + + for row in rows { + let metadata = row?; + existing_files.insert(metadata.path.clone(), metadata); + } + + Ok(existing_files) +} + +/// Analyze files and determine which need updates vs inserts +pub fn analyze_files_for_batch_update( + entries: &[DirEntry], + existing_files: &HashMap +) -> BatchUpdate { + let mut files_to_update = Vec::new(); + let mut files_to_insert = Vec::new(); + + for entry in entries { + let meta = match entry.metadata() { + Ok(m) if !m.is_dir() => m, + _ => continue, + }; + + let fpath = match entry.path().canonicalize() { + Ok(fp) => fp.to_string_lossy().to_string(), + Err(_) => continue, + }; + + let fmodified = match meta.modified() + .ok() + .and_then(|m| m.duration_since(UNIX_EPOCH).ok().map(|d| d.as_secs())) { + Some(time) => time, + None => continue, + }; + + if let Some(existing_metadata) = existing_files.get(&fpath) { + // File exists in database, check if modification date changed + if existing_metadata.moddate != fmodified { + files_to_update.push((entry.clone(), existing_metadata.clone())); + } + // If moddate is same, skip processing this file entirely + } else { + // New file, needs to be inserted + files_to_insert.push(entry.clone()); + } + } + + BatchUpdate { + files_to_update, + files_to_insert, + } +} + /// Get a hash of a file by reading the first and last HASHLEN bytes of the file fn get_file_hash(size: u64, path: OsString) -> Result, std::io::Error> { let mut hasher = Sha256::new(); @@ -59,7 +138,244 @@ fn get_file_hash(size: u64, path: OsString) -> Result, std::io::Error> { Ok(hasher.finalize().to_vec()) } +/// Process updated files in batch with transaction +pub fn process_batch_updates( + conn_mutex: &Arc>, + files_to_update: &[(DirEntry, FileMetadata)], + stop_flag: &Arc>, + status_callback: Option> +) -> Result<(), String> { + if files_to_update.is_empty() { + return Ok(()); + } + const BATCH_SIZE: usize = 1000; + let total_files = files_to_update.len(); + + // Process files in batches of BATCH_SIZE + for (batch_idx, batch) in files_to_update.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, (entry, _old_metadata)) 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 with current file + if let Some(ref callback) = status_callback { + let filename = entry.path().file_name() + .and_then(|n| n.to_str()) + .unwrap_or("unknown"); + callback(&format!("Updating file {}/{}: {}", global_index, total_files, filename)); + } + 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()) + .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 <= 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() > MAXIMUM_TEXT_SIZE { + file_string[..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() > MAXIMUM_TEXT_SIZE { + extracted_text[..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))?; + } + } + } + } + } + + tx.commit().map_err(|e| format!("Failed to commit transaction: {}", e))?; + } + + Ok(()) +} + +/// Process new files in batch with transaction +pub fn process_batch_inserts( + conn_mutex: &Arc>, + files_to_insert: &[DirEntry], + stop_flag: &Arc>, + status_callback: Option> +) -> Result<(), String> { + if files_to_insert.is_empty() { + return Ok(()); + } + + const BATCH_SIZE: usize = 1000; + let total_files = files_to_insert.len(); + + // Process files in batches of BATCH_SIZE + for (batch_idx, batch) in files_to_insert.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, entry) 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 with current file + if let Some(ref callback) = status_callback { + let filename = entry.path().file_name() + .and_then(|n| n.to_str()) + .unwrap_or("unknown"); + callback(&format!("Indexing file {}/{}: {}", global_index, total_files, filename)); + } + 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()) + .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 <= 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() > MAXIMUM_TEXT_SIZE { + file_string[..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() > MAXIMUM_TEXT_SIZE { + extracted_text[..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))?; + } + } + } + } + } + + tx.commit().map_err(|e| format!("Failed to commit transaction: {}", e))?; + } + + Ok(()) +} pub fn process_entry(conn_mutex: &Arc>, entry: DirEntry) { let meta = entry.metadata().unwrap(); diff --git a/src/frontend.rs b/src/frontend.rs index 4606895..1a89214 100644 --- a/src/frontend.rs +++ b/src/frontend.rs @@ -60,7 +60,7 @@ pub fn App(props: AppProps) -> Element { ) } } - IndexingStatus::Stopping => "Stopping...".to_string(), + IndexingStatus::Stopping => "Indexing Stopped".to_string(), IndexingStatus::Error(ref e) => format!("Error: {}", e), }; status_text.set(status_str); @@ -107,7 +107,7 @@ pub fn App(props: AppProps) -> Element { ) } } - IndexingStatus::Stopping => "Stopping...".to_string(), + IndexingStatus::Stopping => "Indexing Stopped".to_string(), IndexingStatus::Error(ref e) => format!("Error: {}", e), }; status_text_clone.set(status_str); diff --git a/src/indexing.rs b/src/indexing.rs index 7f58965..cabac80 100644 --- a/src/indexing.rs +++ b/src/indexing.rs @@ -1,10 +1,10 @@ use std::sync::{Arc, Mutex, mpsc}; use std::thread; -use std::time::{Duration, Instant}; +use std::time::Instant; use walkdir::WalkDir; use rusqlite::Connection; -use crate::file_handling; +use crate::file_handling::{load_existing_files, analyze_files_for_batch_update, process_batch_updates, process_batch_inserts}; #[derive(Debug, Clone)] pub enum IndexingStatus { @@ -69,6 +69,7 @@ impl IndexingService { fn indexing_thread(status: Arc>, command_rx: mpsc::Receiver) { let stop_flag = Arc::new(Mutex::new(false)); + let mut indexing_handle: Option> = None; while let Ok(command) = command_rx.recv() { match command { @@ -77,6 +78,11 @@ impl IndexingService { continue; // Already running } + // Join any previous indexing thread + if let Some(handle) = indexing_handle.take() { + let _ = handle.join(); + } + *stop_flag.lock().unwrap() = false; *status.lock().unwrap() = IndexingStatus::Running { files_processed: 0, @@ -85,12 +91,22 @@ impl IndexingService { start_time: Instant::now(), }; - // Run indexing - if let Err(e) = Self::run_indexing(&status, &path, &db_path, &stop_flag) { - *status.lock().unwrap() = IndexingStatus::Error(e); - } else { - *status.lock().unwrap() = IndexingStatus::Idle; - } + // Run indexing in a separate thread + let status_clone = status.clone(); + let stop_flag_clone = stop_flag.clone(); + let path_owned = path.clone(); + let db_path_owned = db_path.clone(); + + indexing_handle = Some(thread::spawn(move || { + if let Err(e) = Self::run_indexing(&status_clone, &path_owned, &db_path_owned, &stop_flag_clone) { + *status_clone.lock().unwrap() = IndexingStatus::Error(e); + } else { + // Only set to Idle if we weren't stopped + if !*stop_flag_clone.lock().unwrap() { + *status_clone.lock().unwrap() = IndexingStatus::Idle; + } + } + })); } IndexingCommand::Stop => { if matches!(*status.lock().unwrap(), IndexingStatus::Running { .. }) { @@ -100,6 +116,11 @@ impl IndexingService { } } } + + // Clean up any remaining indexing thread + if let Some(handle) = indexing_handle { + let _ = handle.join(); + } } fn run_indexing( @@ -137,14 +158,23 @@ impl IndexingService { ) .map_err(|e| format!("Failed to create searchabletext table: {}", e))?; + // Load existing files from database for incremental indexing + let existing_files = { + let conn_ref = &conn; + load_existing_files(conn_ref) + .map_err(|e| format!("Failed to load existing files: {}", e))? + }; + let conn_mutex = Arc::new(Mutex::new(conn)); - // Count total files for progress tracking - let total_file_count = WalkDir::new(path) - .into_iter() + // Collect all file entries + let walker = WalkDir::new(path).into_iter(); + let entries: Vec<_> = walker .filter_map(|entry| entry.ok()) .filter(|entry| !entry.metadata().map(|m| m.is_dir()).unwrap_or(true)) - .count(); + .collect(); + + let total_file_count = entries.len(); // Update status with total file count if let Ok(mut status_guard) = status.lock() { @@ -153,34 +183,117 @@ impl IndexingService { } } - - // Process files with periodic status updates - let walker = WalkDir::new(path).into_iter(); + // Analyze which files need updates vs inserts + let batch_update = analyze_files_for_batch_update(&entries, &existing_files); - // Use a custom parallel iterator that checks for stop condition - let entries: Vec<_> = walker.filter_map(|entry| entry.ok()).collect(); + let total_work = batch_update.files_to_update.len() + batch_update.files_to_insert.len(); - for (i, entry) in entries.iter().enumerate() { - if *stop_flag.lock().unwrap() { - return Ok(()); + // 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 { + *total_files = Some(total_work); } + } - // Update current file in status + let mut work_completed = 0; + + // 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, ref mut files_processed, .. } = *status_guard { - *current_file = Some(entry.path().to_string_lossy().to_string()); - *files_processed = i; + if let IndexingStatus::Running { ref mut current_file, .. } = *status_guard { + *current_file = Some(format!("Updating {} modified files...", batch_update.files_to_update.len())); } } - // Process the entry - file_handling::process_entry(&conn_mutex, entry.clone()); + // Create status callback to update current file and progress + let status_clone = status.clone(); + let status_callback = Box::new(move |file_status: &str| { + if let Ok(mut status_guard) = status_clone.lock() { + if let IndexingStatus::Running { ref mut current_file, ref mut files_processed, .. } = *status_guard { + *current_file = Some(file_status.to_string()); + // Extract the current count from the status string + if let Some(slash_pos) = file_status.find('/') { + if let Some(space_pos) = file_status.rfind(' ') { + if space_pos + 1 < slash_pos { + if let Ok(current) = file_status[space_pos + 1..slash_pos].parse::() { + *files_processed = work_completed + current; + } + } + } + } + } + } + }); - // Throttle status updates and check for stop signal more frequently - if i % 10 == 0 { - thread::sleep(Duration::from_millis(1)); - if *stop_flag.lock().unwrap() { - return Ok(()); + if let Err(e) = process_batch_updates(&conn_mutex, &batch_update.files_to_update, &stop_flag, Some(status_callback)) { + 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 { + *files_processed = work_completed; + } + } + } + + // Check for stop signal + if *stop_flag.lock().unwrap() { + if let Ok(mut status_guard) = status.lock() { + *status_guard = IndexingStatus::Idle; + } + return Ok(()); + } + + // 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 { + *current_file = Some(format!("Indexing {} new files...", batch_update.files_to_insert.len())); + } + } + + // Create status callback to update current file and progress + let status_clone = status.clone(); + let base_work_completed = work_completed; + let status_callback = Box::new(move |file_status: &str| { + if let Ok(mut status_guard) = status_clone.lock() { + if let IndexingStatus::Running { ref mut current_file, ref mut files_processed, .. } = *status_guard { + *current_file = Some(file_status.to_string()); + // Extract the current count from the status string + if let Some(slash_pos) = file_status.find('/') { + if let Some(space_pos) = file_status.rfind(' ') { + if space_pos + 1 < slash_pos { + if let Ok(current) = file_status[space_pos + 1..slash_pos].parse::() { + *files_processed = base_work_completed + current; + } + } + } + } + } + } + }); + + if let Err(e) = process_batch_inserts(&conn_mutex, &batch_update.files_to_insert, &stop_flag, Some(status_callback)) { + 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 { + *files_processed = work_completed; + } + } + } + + // If no incremental work was needed, show completion status + 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()); + *files_processed = total_file_count; } } } @@ -194,3 +307,4 @@ impl Default for IndexingService { Self::new() } } +