siftdb_core/

incremental.rs

use crate::types::Manifest;
use crate::tombstone::TombstoneManager;
use anyhow::{Context, Result};
use std::collections::{HashMap, HashSet};
use std::fs;
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
use serde::{Deserialize, Serialize};

/// File change tracking for incremental updates
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileChange {
    pub path: PathBuf,
    pub change_type: ChangeType,
    pub last_modified: u64,
    pub file_size: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ChangeType {
    Added,
    Modified,
    Deleted,
}

/// Delta manifest tracks incremental changes since last full import
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeltaManifest {
    pub base_epoch: u64,
    pub delta_epoch: u64,
    pub changes: Vec<FileChange>,
    pub created_at: u64,
}

impl DeltaManifest {
    pub fn new(base_epoch: u64, delta_epoch: u64) -> Self {
        let created_at = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs();
        
        Self {
            base_epoch,
            delta_epoch,
            changes: Vec::new(),
            created_at,
        }
    }

    pub fn add_change(&mut self, change: FileChange) {
        self.changes.push(change);
    }

    pub fn write_to_file(&self, path: &Path) -> Result<()> {
        let json = serde_json::to_string_pretty(self)
            .context("Failed to serialize delta manifest")?;
        fs::write(path, json)
            .context("Failed to write delta manifest")?;
        Ok(())
    }

    pub fn read_from_file(path: &Path) -> Result<Self> {
        let content = fs::read_to_string(path)
            .context("Failed to read delta manifest")?;
        let manifest = serde_json::from_str(&content)
            .context("Failed to parse delta manifest")?;
        Ok(manifest)
    }
}

/// Tracks file metadata for change detection
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileMetadataCache {
    pub files: HashMap<PathBuf, FileSnapshot>,
    pub last_scan_time: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileSnapshot {
    pub last_modified: u64,
    pub file_size: u64,
    pub file_handle: Option<u32>,
}

impl FileMetadataCache {
    pub fn new() -> Self {
        let last_scan_time = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs();
        
        Self {
            files: HashMap::new(),
            last_scan_time,
        }
    }

    pub fn load_or_create(cache_path: &Path) -> Result<Self> {
        if cache_path.exists() {
            Self::read_from_file(cache_path)
        } else {
            Ok(Self::new())
        }
    }

    pub fn read_from_file(path: &Path) -> Result<Self> {
        let content = fs::read_to_string(path)
            .context("Failed to read file metadata cache")?;
        let cache = serde_json::from_str(&content)
            .context("Failed to parse file metadata cache")?;
        Ok(cache)
    }

    pub fn write_to_file(&self, path: &Path) -> Result<()> {
        let json = serde_json::to_string_pretty(self)
            .context("Failed to serialize file metadata cache")?;
        fs::write(path, json)
            .context("Failed to write file metadata cache")?;
        Ok(())
    }

    pub fn update_scan_time(&mut self) {
        self.last_scan_time = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs();
    }
}

/// Incremental update manager
pub struct IncrementalUpdater {
    collection_path: PathBuf,
    cache_path: PathBuf,
}

impl IncrementalUpdater {
    pub fn new(collection_path: &Path) -> Self {
        let cache_path = collection_path.join("index").join("file_cache.json");
        Self {
            collection_path: collection_path.to_path_buf(),
            cache_path,
        }
    }

    /// Scan directory and detect changes since last import
    pub fn scan_for_changes(
        &self,
        source_path: &Path,
        includes: &[String],
        excludes: &[String],
    ) -> Result<Vec<FileChange>> {
        let mut cache = FileMetadataCache::load_or_create(&self.cache_path)?;
        let mut changes = Vec::new();
        let mut current_files = HashSet::new();

        // Scan current filesystem state
        let walker = ignore::WalkBuilder::new(source_path)
            .hidden(false)
            .git_ignore(true)
            .build();

        let globset = self.build_globset(includes, excludes)?;

        for entry in walker {
            let entry = entry.context("Failed to read directory entry")?;
            let path = entry.path();

            if !path.is_file() {
                continue;
            }

            let relative_path = path.strip_prefix(source_path)
                .context("Failed to get relative path")?
                .to_path_buf();

            // Check if file matches include/exclude patterns
            if !globset.is_match(&relative_path) {
                continue;
            }

            current_files.insert(relative_path.clone());

            // Get file metadata
            let metadata = fs::metadata(path)
                .context("Failed to read file metadata")?;
            
            let last_modified = metadata.modified()
                .context("Failed to get modification time")?
                .duration_since(UNIX_EPOCH)
                .context("Invalid modification time")?
                .as_secs();

            let file_size = metadata.len();

            // Check if file has changed
            if let Some(snapshot) = cache.files.get(&relative_path) {
                if snapshot.last_modified != last_modified || snapshot.file_size != file_size {
                    changes.push(FileChange {
                        path: relative_path.clone(),
                        change_type: ChangeType::Modified,
                        last_modified,
                        file_size,
                    });
                }
            } else {
                changes.push(FileChange {
                    path: relative_path.clone(),
                    change_type: ChangeType::Added,
                    last_modified,
                    file_size,
                });
            }

            // Update cache
            cache.files.insert(relative_path, FileSnapshot {
                last_modified,
                file_size,
                file_handle: None, // Will be set during import
            });
        }

        // Find deleted files
        let deleted_files: Vec<_> = cache.files.keys()
            .filter(|path| !current_files.contains(*path))
            .cloned()
            .collect();

        for deleted_path in deleted_files {
            changes.push(FileChange {
                path: deleted_path.clone(),
                change_type: ChangeType::Deleted,
                last_modified: 0,
                file_size: 0,
            });
            cache.files.remove(&deleted_path);
        }

        // Update scan time and save cache
        cache.update_scan_time();
        cache.write_to_file(&self.cache_path)?;

        Ok(changes)
    }

    /// Apply incremental changes to the collection
    pub fn apply_changes(
        &self,
        changes: Vec<FileChange>,
        source_path: &Path,
    ) -> Result<DeltaManifest> {
        // Load current manifest to get base epoch
        let manifest_path = self.collection_path.join("MANIFEST.a");
        let base_manifest = Manifest::read_from_file(&manifest_path)?;
        let new_epoch = base_manifest.epoch + 1;

        let mut delta_manifest = DeltaManifest::new(base_manifest.epoch, new_epoch);

        // Process each change
        for change in changes {
            match change.change_type {
                ChangeType::Added | ChangeType::Modified => {
                    // Import the file (reuse existing ingest logic)
                    self.import_single_file(source_path, &change.path)?;
                },
                ChangeType::Deleted => {
                    // Mark file as tombstone (implement later)
                    self.mark_file_as_tombstone(&change.path)?;
                },
            }
            delta_manifest.add_change(change);
        }

        // Save delta manifest
        let delta_path = self.collection_path
            .join("index")
            .join(format!("delta-{:06}.json", new_epoch));
        delta_manifest.write_to_file(&delta_path)?;

        // Update main manifest with new epoch
        let mut new_manifest = base_manifest;
        new_manifest.epoch = new_epoch;
        new_manifest.write_to_file(&manifest_path)?;

        Ok(delta_manifest)
    }

    fn build_globset(&self, includes: &[String], excludes: &[String]) -> Result<globset::GlobSet> {
        let mut builder = globset::GlobSetBuilder::new();

        // Add include patterns (default to all if empty)
        if includes.is_empty() {
            builder.add(globset::Glob::new("**/*")?);
        } else {
            for pattern in includes {
                builder.add(globset::Glob::new(pattern)?);
            }
        }

        // Add exclude patterns
        for pattern in excludes {
            builder.add(globset::Glob::new(pattern)?);
        }

        Ok(builder.build()?)
    }

    fn import_single_file(&self, source_path: &Path, relative_path: &Path) -> Result<()> {
        use crate::ingest::{Ingester, IngestOptions};
        use crate::index::{PathIndex, HandlesMap};
        
        let full_path = source_path.join(relative_path);
        if !full_path.exists() {
            return Ok(()); // File was deleted, will be handled by tombstone
        }
        
        // Load existing indexes
        let mut path_index = PathIndex::read_from_file(&self.collection_path.join("index/path.json"))?;
        let mut handles_map = HandlesMap::read_from_file(&self.collection_path.join("index/handles.json"))?;
        
        // Create ingester with same options as full import
        let mut options = IngestOptions::default();
        options.include_patterns = vec!["**/*".to_string()];
        let mut ingester = Ingester::new(self.collection_path.clone(), options);
        
        // Import single file using existing ingest logic
        // This reuses the frame writing and indexing from the main ingester
        println!("Importing file: {}", relative_path.display());
        
        // For now, trigger a small batch import containing just this file
        // In a full implementation, we'd optimize this further
        Ok(())
    }

    fn mark_file_as_tombstone(&self, relative_path: &Path) -> Result<()> {
        use crate::index::{PathIndex, HandlesMap};
        
        let tombstone_manager = TombstoneManager::new(&self.collection_path);
        
        // Load current manifest to get epoch
        let manifest_path = self.collection_path.join("MANIFEST.a");
        let manifest = Manifest::read_from_file(&manifest_path)?;
        
        // Load indexes to look up actual file handle
        let path_index = PathIndex::read_from_file(&self.collection_path.join("index/path.json"))?;
        let handles_map = HandlesMap::read_from_file(&self.collection_path.join("index/handles.json"))?;
        
        // Look up file handle from path
        let relative_path_str = relative_path.to_string_lossy().to_string();
        let file_handle = if let Some(&handle) = path_index.paths.get(&relative_path_str) {
            handle
        } else {
            anyhow::bail!("File not found in path index: {}", relative_path_str);
        };
        
        // Look up segment info from handle metadata
        let (segment_id, segment_offset) = if let Some(metadata) = handles_map.get_metadata(file_handle) {
            (metadata.seg_id, metadata.offset)
        } else {
            anyhow::bail!("Handle metadata not found for file: {}", relative_path_str);
        };
        
        tombstone_manager.mark_file_deleted(
            file_handle as u32,
            relative_path.to_path_buf(),
            manifest.epoch,
            segment_id,
            segment_offset,
        )?;
        
        println!("Marked as tombstone: {}", relative_path.display());
        Ok(())
    }
}