jeremy/quick_search
jeremy/quick_search
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions

Added graceful shutdown and closing of DB connections

Browse source
This commit is contained in:
Jeremy Karst 2025-09-11 18:36:42 -04:00
parent 4eb3130c58
commit c39dd9cc03
8 changed files with 505 additions and 191 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
Cargo.lock generated
View file

@ -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",

3
Cargo.toml
View file

@ -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"

3
config_example.toml
View file

@ -11,3 +11,6 @@ maximum_text_size = 524288
maximum_file_size = 52428800
# Number of files to process in each batch
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"

2
src/config.rs
View file

@ -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(),
},
}
}

354
src/file_handling.rs
View file

@ -26,7 +26,7 @@ pub struct BatchUpdate {
pub files_to_insert: Vec<DirEntry>,
}
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<Vec<u8>, std::io::Error> {
let mut hasher = Sha256::new();
@ -136,8 +151,8 @@ fn get_file_hash(size: u64, path: OsString, hash_length: usize) -> Result<Vec<u8
Ok(hasher.finalize().to_vec())
}
/// Process updated files in batch with transaction
pub fn process_batch_updates(
/// Process updated files in batch with transaction - files table only (no text extraction)
pub fn process_batch_updates_files_only(
conn_mutex: &Arc<Mutex<Connection>>,
files_to_update: &[(DirEntry, FileMetadata)],
stop_flag: &Arc<Mutex<bool>>,
@ -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<Mutex<Connection>>,
files_to_insert: &[DirEntry],
stop_flag: &Arc<Mutex<bool>>,
@ -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<Mutex<Connection>>,
stop_flag: &Arc<Mutex<bool>>,
status_callback: Option<Box<dyn Fn(&str) + Send + Sync>>,
progress_callback: Option<Box<dyn Fn(usize) + Send + Sync>>,
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::<Result<Vec<_>, _>>()
.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))?;

48
src/frontend.rs
View file

@ -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,

205
src/indexing.rs
View file

@ -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<usize>,
current_file: Option<String>,
start_time: Instant,
},
RunningTextIndex {
files_processed: usize,
total_files: Option<usize>,
current_file: Option<String>,
@ -30,9 +36,11 @@ pub enum IndexingCommand {
Stop,
}
#[derive(Debug)]
pub struct IndexingService {
status: Arc<Mutex<IndexingStatus>>,
command_tx: mpsc::Sender<IndexingCommand>,
db_connection: Arc<Mutex<Option<Arc<Mutex<Connection>>>>>,
_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<Option<Vec<String>>, 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<Mutex<IndexingStatus>>, command_rx: mpsc::Receiver<IndexingCommand>) {
fn indexing_thread(
status: Arc<Mutex<IndexingStatus>>,
command_rx: mpsc::Receiver<IndexingCommand>,
db_connection: Arc<Mutex<Option<Arc<Mutex<Connection>>>>>
) {
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 {
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<Mutex<bool>>,
config: &Config,
db_connection: &Arc<Mutex<Option<Arc<Mutex<Connection>>>>>,
) -> 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 (
@ -254,6 +318,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();
let entries: Vec<_> = walker
@ -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<Option<Vec<String>>, 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<String> = None;
let mut stored_indexing_path: Option<String> = None;
let mut stored_tokenize: Option<String> = 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))
@ -465,10 +594,22 @@ impl IndexingService {
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()

22
src/main.rs
View file

@ -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<Arc<indexing::IndexingService>> = 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 {