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};
#[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,
}
#[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)
}
}
#[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();
}
}
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,
}
}
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();
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();
if !globset.is_match(&relative_path) {
continue;
}
current_files.insert(relative_path.clone());
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();
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,
});
}
cache.files.insert(relative_path, FileSnapshot {
last_modified,
file_size,
file_handle: None, });
}
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);
}
cache.update_scan_time();
cache.write_to_file(&self.cache_path)?;
Ok(changes)
}
pub fn apply_changes(
&self,
changes: Vec<FileChange>,
source_path: &Path,
) -> Result<DeltaManifest> {
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);
for change in changes {
match change.change_type {
ChangeType::Added | ChangeType::Modified => {
self.import_single_file(source_path, &change.path)?;
},
ChangeType::Deleted => {
self.mark_file_as_tombstone(&change.path)?;
},
}
delta_manifest.add_change(change);
}
let delta_path = self.collection_path
.join("index")
.join(format!("delta-{:06}.json", new_epoch));
delta_manifest.write_to_file(&delta_path)?;
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();
if includes.is_empty() {
builder.add(globset::Glob::new("**/*")?);
} else {
for pattern in includes {
builder.add(globset::Glob::new(pattern)?);
}
}
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(()); }
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"))?;
let mut options = IngestOptions::default();
options.include_patterns = vec!["**/*".to_string()];
let mut ingester = Ingester::new(self.collection_path.clone(), options);
println!("Importing file: {}", relative_path.display());
Ok(())
}
fn mark_file_as_tombstone(&self, relative_path: &Path) -> Result<()> {
use crate::index::{PathIndex, HandlesMap};
let tombstone_manager = TombstoneManager::new(&self.collection_path);
let manifest_path = self.collection_path.join("MANIFEST.a");
let manifest = Manifest::read_from_file(&manifest_path)?;
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"))?;
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);
};
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(())
}
}