use std::path::Path;
use std::fs;
use std::io::Read;
use std::collections::HashMap;
use ignore::WalkBuilder;
use anyhow::Result;
use crate::types::*;
use crate::storage::{SegmentWriter, generate_line_table};
use crate::index::{PathIndex, HandlesMap};
use crate::locking::{SWMRLockManager, WriteLock};
pub struct IngestOptions {
pub include_patterns: Vec<String>,
pub exclude_patterns: Vec<String>,
pub max_file_bytes: u64,
pub binary_ratio_threshold: f32,
}
impl Default for IngestOptions {
fn default() -> Self {
Self {
include_patterns: vec!["**/*".to_string()],
exclude_patterns: vec![
"**/target/**".to_string(),
"**/node_modules/**".to_string(),
"**/.git/**".to_string(),
"**/build/**".to_string(),
"**/dist/**".to_string(),
],
max_file_bytes: 10 * 1024 * 1024, binary_ratio_threshold: 0.3, }
}
}
pub struct Ingester {
collection_path: std::path::PathBuf,
options: IngestOptions,
lock_manager: SWMRLockManager,
}
impl Ingester {
pub fn new(collection_path: std::path::PathBuf, options: IngestOptions) -> Self {
let lock_manager = SWMRLockManager::new(&collection_path);
Self {
collection_path,
options,
lock_manager,
}
}
pub fn ingest_from_fs(&mut self, source_path: &Path) -> Result<IngestStats> {
self.ingest_from_fs_with_lock_config(source_path, 300, "sift-import".to_string())
}
pub fn ingest_from_fs_with_lock_config(&mut self, source_path: &Path, timeout_secs: u64, holder_info: String) -> Result<IngestStats> {
let lock_manager = SWMRLockManager::new(&self.collection_path);
let _write_lock = lock_manager.acquire_write_lock(timeout_secs, holder_info)?;
self.ingest_from_fs_impl(source_path)
}
fn ingest_from_fs_impl(&mut self, source_path: &Path) -> Result<IngestStats> {
let mut stats = IngestStats::default();
let mut ingested_content = HashMap::new();
let mut path_mappings = HashMap::new();
let mut handle_metadata = HashMap::new();
let mut path_index = if self.collection_path.join("index/path.json").exists() {
PathIndex::read_from_file(&self.collection_path.join("index/path.json"))?
} else {
PathIndex::new()
};
let mut handles_map = if self.collection_path.join("index/handles.json").exists() {
HandlesMap::read_from_file(&self.collection_path.join("index/handles.json"))?
} else {
HandlesMap::new()
};
let store_path = self.collection_path.join("store");
let seg_id = self.find_next_segment_id(&store_path)?;
let mut writer = SegmentWriter::new(&store_path, seg_id)?;
let walker = WalkBuilder::new(source_path)
.hidden(false) .git_ignore(true)
.git_global(true)
.git_exclude(true)
.build();
for entry in walker {
let entry = entry?;
let path = entry.path();
if path.is_dir() {
continue;
}
if !self.should_include_file(path)? {
stats.skipped += 1;
continue;
}
let content = match fs::read(path) {
Ok(content) => content,
Err(e) => {
eprintln!("Warning: Failed to read {}: {}", path.display(), e);
stats.errors += 1;
continue;
}
};
if content.len() > self.options.max_file_bytes as usize {
stats.skipped += 1;
continue;
}
if self.is_binary(&content) {
stats.skipped += 1;
continue;
}
let relative_path = path.strip_prefix(source_path)
.unwrap_or(path)
.to_string_lossy()
.to_string();
let lang = detect_language(path);
let line_table = generate_line_table(&content);
let header = FileHeader::new(&content, &line_table, lang);
let frame = Frame {
header,
content: content.clone(),
line_table,
};
let handle = path_index.add_path(relative_path.clone());
let metadata = writer.write_frame(&frame)?;
handles_map.add_handle(handle, metadata.clone());
ingested_content.insert(handle, content);
path_mappings.insert(relative_path, handle);
handle_metadata.insert(handle, metadata);
stats.ingested += 1;
if stats.ingested % 100 == 0 {
println!("Ingested {} files...", stats.ingested);
}
}
let mut path_index = PathIndex::new();
for (path, handle) in path_mappings {
path_index.paths.insert(path, handle);
}
let mut handles_map = HandlesMap::new();
for (handle, metadata) in handle_metadata {
handles_map.add_handle(handle, metadata);
}
path_index.write_to_file(&self.collection_path.join("index/path.json"))?;
handles_map.write_to_file(&self.collection_path.join("index/handles.json"))?;
println!("Building inverted index for O(1) search...");
let mut file_contents = HashMap::new();
for (file_handle, content) in &ingested_content {
if let Ok(content_str) = String::from_utf8(content.clone()) {
file_contents.insert(*file_handle as u32, content_str);
}
}
if !file_contents.is_empty() {
let inverted_index = crate::inverted_index::InvertedIndex::build_from_content(
file_contents,
&self.collection_path.join("index/terms.fst"),
&self.collection_path.join("index/posting_lists.json")
)?;
println!("✓ Inverted index built with {} terms", inverted_index.term_count());
}
println!(
"Ingestion complete: {} files ingested, {} skipped, {} errors",
stats.ingested, stats.skipped, stats.errors
);
Ok(stats)
}
fn should_include_file(&self, path: &Path) -> Result<bool> {
let path_str = path.to_string_lossy();
for pattern in &self.options.exclude_patterns {
if self.glob_match(pattern, &path_str) {
return Ok(false);
}
}
for pattern in &self.options.include_patterns {
if self.glob_match(pattern, &path_str) {
return Ok(true);
}
}
Ok(false)
}
fn is_binary(&self, content: &[u8]) -> bool {
if content.is_empty() {
return false;
}
let mut non_printable = 0;
for &byte in content.iter().take(1024) { if byte < 32 && byte != 9 && byte != 10 && byte != 13 {
non_printable += 1;
}
}
let ratio = non_printable as f32 / content.len().min(1024) as f32;
ratio > self.options.binary_ratio_threshold
}
fn glob_match(&self, pattern: &str, text: &str) -> bool {
if pattern == "**/*" {
return true;
}
if pattern.starts_with("**/") {
let suffix = &pattern[3..];
if suffix.starts_with("*.") {
let ext = &suffix[1..]; return text.ends_with(ext);
} else {
return text.ends_with(suffix);
}
}
if pattern.starts_with("**/") && pattern.ends_with("/**") {
let dir_name = &pattern[3..pattern.len()-3];
return text.contains(&format!("/{}/", dir_name)) ||
text.starts_with(&format!("{}/", dir_name));
}
if pattern.ends_with("/**") {
let prefix = &pattern[..pattern.len()-3];
return text.starts_with(prefix);
}
if pattern.starts_with("*.") {
let ext = &pattern[1..]; return text.ends_with(ext);
}
if !pattern.contains('/') && !pattern.contains('*') {
if let Some(filename) = text.split('/').last() {
if filename == pattern {
return true;
}
}
}
if pattern.contains('*') {
return self.wildcard_match(pattern, text);
}
pattern == text
}
fn wildcard_match(&self, pattern: &str, text: &str) -> bool {
let pattern_chars: Vec<char> = pattern.chars().collect();
let text_chars: Vec<char> = text.chars().collect();
self.match_recursive(&pattern_chars, &text_chars, 0, 0)
}
fn match_recursive(&self, pattern: &[char], text: &[char], p_idx: usize, t_idx: usize) -> bool {
if p_idx == pattern.len() {
return t_idx == text.len();
}
if pattern[p_idx] == '*' {
for i in t_idx..=text.len() {
if self.match_recursive(pattern, text, p_idx + 1, i) {
return true;
}
}
false
} else if t_idx < text.len() && (pattern[p_idx] == text[t_idx] || pattern[p_idx] == '?') {
self.match_recursive(pattern, text, p_idx + 1, t_idx + 1)
} else {
false
}
}
fn find_next_segment_id(&self, store_path: &Path) -> Result<u32> {
let mut max_id = 0;
if store_path.exists() {
for entry in fs::read_dir(store_path)? {
let entry = entry?;
let name = entry.file_name();
let name_str = name.to_string_lossy();
if name_str.starts_with("seg-") && name_str.ends_with(".sift") {
if let Some(id_str) = name_str.strip_prefix("seg-").and_then(|s| s.strip_suffix(".sift")) {
if let Ok(id) = id_str.parse::<u32>() {
max_id = max_id.max(id);
}
}
}
}
}
Ok(max_id + 1)
}
}
#[derive(Debug, Clone)]
pub struct IngestStats {
pub ingested: u64,
pub skipped: u64,
pub errors: u64,
}
impl IngestStats {
pub fn new() -> Self {
Self {
ingested: 0,
skipped: 0,
errors: 0,
}
}
}
impl Default for IngestStats {
fn default() -> Self {
Self::new()
}
}