siftdb_core/

ingest.rs

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, // 10MB
            binary_ratio_threshold: 0.3, // 30% non-printable chars = binary
        }
    }
}

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,
        }
    }
    
    /// Ingest from filesystem with default lock configuration
    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())
    }
    
    /// Ingest from filesystem with custom lock configuration
    pub fn ingest_from_fs_with_lock_config(&mut self, source_path: &Path, timeout_secs: u64, holder_info: String) -> Result<IngestStats> {
        // Acquire write lock before starting ingestion - temporarily simplified to avoid borrowing conflicts
        // TODO: Properly integrate lock management with ingestion lifecycle
        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)
    }
    
    /// Internal implementation of ingestion (assumes lock is already held)
    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();
        
        // Load existing indexes or create new ones
        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()
        };
        
        // Create segment writer
        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)?;
        
        // Walk directory with ignore patterns
        let walker = WalkBuilder::new(source_path)
            .hidden(false) // Include hidden files
            .git_ignore(true)
            .git_global(true)
            .git_exclude(true)
            .build();
        
        for entry in walker {
            let entry = entry?;
            let path = entry.path();
            
            // Skip directories
            if path.is_dir() {
                continue;
            }
            
            // Apply filters
            if !self.should_include_file(path)? {
                stats.skipped += 1;
                continue;
            }
            
            // Read file content
            let content = match fs::read(path) {
                Ok(content) => content,
                Err(e) => {
                    eprintln!("Warning: Failed to read {}: {}", path.display(), e);
                    stats.errors += 1;
                    continue;
                }
            };
            
            // Check file size
            if content.len() > self.options.max_file_bytes as usize {
                stats.skipped += 1;
                continue;
            }
            
            // Check if binary
            if self.is_binary(&content) {
                stats.skipped += 1;
                continue;
            }
            
            // Generate relative path from source
            let relative_path = path.strip_prefix(source_path)
                .unwrap_or(path)
                .to_string_lossy()
                .to_string();
            
            // Detect language
            let lang = detect_language(path);
            
            // Generate line table
            let line_table = generate_line_table(&content);
            
            // Create frame
            let header = FileHeader::new(&content, &line_table, lang);
            let frame = Frame {
                header,
                content: content.clone(),
                line_table,
            };
            
            // Write frame and get handle
            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);
            }
        }
        
        // Build and save indexes
        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"))?;
        
        // Build inverted index from file contents for O(1) search
        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();
        
        // Check exclude patterns first
        for pattern in &self.options.exclude_patterns {
            if self.glob_match(pattern, &path_str) {
                return Ok(false);
            }
        }
        
        // Check include patterns
        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) { // Check first 1KB
            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 {
        // Simple glob matching for MVP
        if pattern == "**/*" {
            return true;
        }
        
        // Handle **/*.ext patterns (recursive file extension matching)
        if pattern.starts_with("**/") {
            let suffix = &pattern[3..];
            if suffix.starts_with("*.") {
                // Extract extension from pattern like "*.rs"
                let ext = &suffix[1..]; // Include the dot
                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);
        }
        
        // Handle simple extension patterns like "*.rs"
        if pattern.starts_with("*.") {
            let ext = &pattern[1..]; // Include the dot
            return text.ends_with(ext);
        }
        
        // Handle exact filename matches (check if pattern matches just the filename)
        if !pattern.contains('/') && !pattern.contains('*') {
            if let Some(filename) = text.split('/').last() {
                if filename == pattern {
                    return true;
                }
            }
        }
        
        // Wildcard matching
        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] == '*' {
            // Try matching 0 or more characters
            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()
    }
}