Fixed resuming from interrupted indexing, and frontend bugs.
This commit is contained in:
parent
3e3369ba27
commit
a85d2261dd
3 changed files with 463 additions and 33 deletions
|
|
@ -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<u8>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct BatchUpdate {
|
||||
pub files_to_update: Vec<(DirEntry, FileMetadata)>,
|
||||
pub files_to_insert: Vec<DirEntry>,
|
||||
}
|
||||
|
||||
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<HashMap<String, FileMetadata>, 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<String, FileMetadata>
|
||||
) -> 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<Vec<u8>, std::io::Error> {
|
||||
let mut hasher = Sha256::new();
|
||||
|
|
@ -59,7 +138,244 @@ fn get_file_hash(size: u64, path: OsString) -> Result<Vec<u8>, std::io::Error> {
|
|||
Ok(hasher.finalize().to_vec())
|
||||
}
|
||||
|
||||
/// Process updated files in batch with transaction
|
||||
pub fn process_batch_updates(
|
||||
conn_mutex: &Arc<Mutex<Connection>>,
|
||||
files_to_update: &[(DirEntry, FileMetadata)],
|
||||
stop_flag: &Arc<Mutex<bool>>,
|
||||
status_callback: Option<Box<dyn Fn(&str) + Send + Sync>>
|
||||
) -> 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<Mutex<Connection>>,
|
||||
files_to_insert: &[DirEntry],
|
||||
stop_flag: &Arc<Mutex<bool>>,
|
||||
status_callback: Option<Box<dyn Fn(&str) + Send + Sync>>
|
||||
) -> 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<Mutex<Connection>>, entry: DirEntry) {
|
||||
let meta = entry.metadata().unwrap();
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
176
src/indexing.rs
176
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<Mutex<IndexingStatus>>, command_rx: mpsc::Receiver<IndexingCommand>) {
|
||||
let stop_flag = Arc::new(Mutex::new(false));
|
||||
let mut indexing_handle: Option<thread::JoinHandle<()>> = 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 {
|
|||
}
|
||||
}
|
||||
|
||||
// Analyze which files need updates vs inserts
|
||||
let batch_update = analyze_files_for_batch_update(&entries, &existing_files);
|
||||
|
||||
// Process files with periodic status updates
|
||||
let walker = WalkDir::new(path).into_iter();
|
||||
let total_work = batch_update.files_to_update.len() + batch_update.files_to_insert.len();
|
||||
|
||||
// Use a custom parallel iterator that checks for stop condition
|
||||
let entries: Vec<_> = walker.filter_map(|entry| entry.ok()).collect();
|
||||
|
||||
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::<usize>() {
|
||||
*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::<usize>() {
|
||||
*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()
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue