use std::collections::{BTreeMap, HashMap};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Instant;
use rayon::prelude::*;
use crate::cache::{ExtractionCache, Fingerprint};
use crate::error::{Diagnostic, Severity};
use crate::extractor::{self, ExtractOptions, ExtractionIdGenerators, InMemorySource};
use crate::graph::GraphBuilder;
use crate::input;
use crate::language::LangId;
use crate::model::{FileExtraction, SnapshotId};
use crate::pipeline::GraphAnalysis;
#[derive(Debug, Clone)]
pub struct Overlay {
pub uri: String,
pub path: PathBuf,
pub text: String,
pub version: i32,
pub lang: LangId,
}
#[derive(Debug, Clone, Default)]
pub struct ChangeSet {
pub files_added: usize,
pub files_removed: usize,
pub files_modified: usize,
pub files_unchanged: usize,
}
pub type ReanalysisOutput = (Vec<Arc<FileExtraction>>, ChangeSet, Vec<Diagnostic>);
pub struct WatchState {
pub(crate) cache: ExtractionCache,
pub(crate) snapshot_counter: u32,
}
impl WatchState {
pub fn new() -> Self {
Self {
cache: ExtractionCache::new(),
snapshot_counter: 0,
}
}
pub fn cache(&self) -> &ExtractionCache {
&self.cache
}
pub fn cache_mut(&mut self) -> &mut ExtractionCache {
&mut self.cache
}
pub(crate) fn next_snapshot_id(&mut self) -> SnapshotId {
self.snapshot_counter += 1;
SnapshotId::new(self.snapshot_counter).expect("snapshot counter exhausted (> u32::MAX)")
}
}
impl Default for WatchState {
fn default() -> Self {
Self::new()
}
}
pub fn reanalyze_extractions(
root: &Path,
languages: Option<&[LangId]>,
overlays: &[Overlay],
state: &mut WatchState,
) -> Result<ReanalysisOutput, crate::Error> {
let files = input::discover_files(root, languages)?;
let overlay_by_path: HashMap<&Path, &Overlay> = overlays
.iter()
.filter(|overlay| overlay.path.starts_with(root))
.map(|overlay| (overlay.path.as_path(), overlay))
.collect();
let mut targets: BTreeMap<PathBuf, LangId> = files.into_iter().collect();
for overlay in overlay_by_path.values() {
targets.entry(overlay.path.clone()).or_insert(overlay.lang);
}
let (current_fingerprints, read_diagnostics): (HashMap<PathBuf, Fingerprint>, Vec<Diagnostic>) =
targets
.par_iter()
.fold(
|| (HashMap::new(), Vec::new()),
|(mut map, mut diags), (path, _)| {
match overlay_by_path.get(path.as_path()) {
Some(overlay) => {
map.insert(path.clone(), Fingerprint::of(overlay.text.as_bytes()));
}
None => match std::fs::read(path) {
Ok(bytes) => {
map.insert(path.clone(), Fingerprint::of(&bytes));
}
Err(err) => diags.push(Diagnostic {
path: path.clone(),
severity: Severity::Error,
message: format!("Failed to read file: {err}"),
source_range: None,
}),
},
}
(map, diags)
},
)
.reduce(
|| (HashMap::new(), Vec::new()),
|(mut m1, mut d1), (m2, d2)| {
m1.extend(m2);
d1.extend(d2);
(m1, d1)
},
);
let mut change_set = ChangeSet::default();
let mut changed_disk: Vec<(PathBuf, LangId)> = Vec::new();
let mut changed_overlays: Vec<&Overlay> = Vec::new();
for (path, lang) in &targets {
let Some(curr_fp) = current_fingerprints.get(path) else {
continue;
};
let changed = match state.cache.fingerprint_of(path) {
Some(cached) if cached == *curr_fp => {
change_set.files_unchanged += 1;
false
}
Some(_) => {
change_set.files_modified += 1;
true
}
None => {
change_set.files_added += 1;
true
}
};
if !changed {
continue;
}
match overlay_by_path.get(path.as_path()) {
Some(overlay) => changed_overlays.push(*overlay),
None => changed_disk.push((path.clone(), *lang)),
}
}
let stale: Vec<PathBuf> = state
.cache
.paths()
.filter(|path| !current_fingerprints.contains_key(*path))
.cloned()
.collect();
if !stale.is_empty() {
change_set.files_removed += stale.len();
for path in &stale {
state.cache.remove(path);
}
}
let max_id = state.cache.max_symbol_id();
#[cfg(feature = "dataflow")]
let max_data_id = state.cache.max_data_node_id();
#[cfg(feature = "dataflow")]
let id_generators = ExtractionIdGenerators::with_starts(max_id + 1, max_data_id + 1);
#[cfg(not(feature = "dataflow"))]
let id_generators = ExtractionIdGenerators::with_symbol_start(max_id + 1);
let options = ExtractOptions {
skip_imports_and_refs: false,
};
let mut new_extractions: Vec<FileExtraction> = if changed_disk.is_empty() {
Vec::new()
} else {
extractor::extract_with_id_gen(&changed_disk, &options, &id_generators).files
};
let mut overlay_diagnostics: Vec<Diagnostic> = Vec::new();
for overlay in &changed_overlays {
match extractor::extract_text_with_id_gen(
InMemorySource {
uri: overlay.uri.as_str(),
text: overlay.text.as_str(),
version: overlay.version,
language: overlay.lang,
},
&options,
&id_generators,
) {
Ok(versioned) => new_extractions.push(versioned.file),
Err(error) => {
let message = error.to_string();
overlay_diagnostics.push(Diagnostic {
path: overlay.path.clone(),
severity: Severity::Error,
message: message.clone(),
source_range: None,
});
new_extractions.push(FileExtraction::failed(
overlay.path.clone(),
overlay.lang,
message,
));
}
}
}
let mut merged: Vec<Arc<FileExtraction>> =
Vec::with_capacity(state.cache.len() + new_extractions.len());
for (path, fp) in ¤t_fingerprints {
if state.cache.fingerprint_of(path) == Some(*fp)
&& let Some(extraction) = state.cache.get(path)
{
merged.push(Arc::clone(extraction));
}
}
for extraction in new_extractions {
let arc = Arc::new(extraction);
if let Some(fp) = current_fingerprints.get(&arc.path) {
state.cache.update(arc.path.clone(), *fp, Arc::clone(&arc));
}
merged.push(arc);
}
merged.sort_by(|a, b| a.path.cmp(&b.path));
let mut diagnostics: Vec<Diagnostic> = merged
.iter()
.flat_map(|file| file.diagnostics.iter().cloned())
.collect();
let mut read_diagnostics = read_diagnostics;
read_diagnostics.sort_by(|a, b| (&a.path, &a.message).cmp(&(&b.path, &b.message)));
diagnostics.extend(read_diagnostics);
diagnostics.extend(overlay_diagnostics);
diagnostics.sort_by(|a, b| (&a.path, &a.message).cmp(&(&b.path, &b.message)));
Ok((merged, change_set, diagnostics))
}
pub fn incremental_reanalyze(
root: &Path,
languages: Option<&[LangId]>,
state: &mut WatchState,
) -> Result<(GraphAnalysis, ChangeSet, Vec<Diagnostic>), crate::Error> {
let started = Instant::now();
let (merged, change_set, mut diagnostics) = reanalyze_extractions(root, languages, &[], state)?;
let snapshot_id = state.next_snapshot_id();
let (graph, scc) = GraphBuilder::from_extractions(&merged, root, snapshot_id, &mut diagnostics);
diagnostics.sort_by(|a, b| (&a.path, &a.message).cmp(&(&b.path, &b.message)));
tracing::info!(
total = merged.len(),
added = change_set.files_added,
removed = change_set.files_removed,
modified = change_set.files_modified,
unchanged = change_set.files_unchanged,
elapsed_ms = started.elapsed().as_millis(),
"Incremental re-analysis complete",
);
let analysis = GraphAnalysis {
graph,
scc,
snapshot_id,
extractions: merged,
};
Ok((analysis, change_set, diagnostics))
}
#[cfg(test)]
mod tests {
use std::io::Write;
use super::*;
fn temp_dir(name: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!("meta_ast_reanalyze_{name}"));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
dir
}
fn write_file(root: &Path, name: &str, content: &str) -> PathBuf {
let path = root.join(name);
let mut file = std::fs::File::create(&path).unwrap();
file.write_all(content.as_bytes()).unwrap();
path
}
fn overlay(root: &Path, name: &str, content: &str) -> Overlay {
let path = root.join(name);
let uri = url::Url::from_file_path(&path)
.expect("absolute path")
.to_string();
Overlay {
uri,
path,
text: content.to_string(),
version: 1,
lang: LangId::Python,
}
}
fn symbol_names(extractions: &[Arc<FileExtraction>]) -> Vec<String> {
extractions
.iter()
.flat_map(|file| file.symbols.iter().map(|s| s.name.clone()))
.collect()
}
#[test]
fn cold_analysis_populates_state() {
let root = temp_dir("cold");
write_file(&root, "a.py", "def foo(): pass\n");
let mut state = WatchState::new();
let (analysis, cs, diags) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert!(diags.is_empty());
assert_eq!(cs.files_added, 1);
assert_eq!(cs.files_unchanged, 0);
assert_eq!(analysis.graph.file_count(), 1);
assert_eq!(analysis.graph.symbol_count(), 1);
assert_eq!(state.cache.extractions.len(), 1);
}
#[test]
fn warm_analysis_reuses_cached_unchanged_files() {
let root = temp_dir("warm");
write_file(&root, "a.py", "def foo(): pass\n");
write_file(&root, "b.py", "def bar(): pass\n");
let mut state = WatchState::new();
let (analysis, cs, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert_eq!(cs.files_added, 2);
assert_eq!(analysis.graph.symbol_count(), 2);
let (analysis2, cs2, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert_eq!(cs2.files_unchanged, 2);
assert_eq!(cs2.files_modified, 0);
assert_eq!(cs2.files_added, 0);
assert_eq!(cs2.files_removed, 0);
assert_eq!(analysis2.graph.symbol_count(), 2);
}
#[test]
fn merged_extractions_stay_path_sorted() {
let root = temp_dir("sorted");
write_file(&root, "b.py", "def bar(): pass\n");
write_file(&root, "a.py", "def foo(): pass\n");
let mut state = WatchState::new();
let (analysis, _, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
let paths: Vec<_> = analysis
.extractions
.iter()
.map(|f| f.path.clone())
.collect();
let mut sorted = paths.clone();
sorted.sort();
assert_eq!(paths, sorted);
}
#[test]
fn detects_modified_file_and_re_extracts() {
let root = temp_dir("mod");
let a = write_file(&root, "a.py", "def original(): pass\n");
write_file(&root, "b.py", "def bar(): pass\n");
let mut state = WatchState::new();
let (analysis, cs, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert_eq!(cs.files_added, 2);
assert_eq!(analysis.graph.symbol_count(), 2);
std::fs::write(&a, "def modified(): pass\ndef extra(): pass\n").unwrap();
let (analysis2, cs2, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert_eq!(cs2.files_unchanged, 1);
assert_eq!(cs2.files_modified, 1);
assert_eq!(analysis2.graph.symbol_count(), 3);
let names: Vec<String> = analysis2
.graph
.symbols()
.map(|(_, s)| s.name.clone())
.collect();
assert!(names.contains(&"modified".to_string()));
assert!(!names.contains(&"original".to_string()));
assert!(names.contains(&"bar".to_string()));
}
#[test]
fn detects_removed_file() {
let root = temp_dir("rem");
let a = write_file(&root, "a.py", "def foo(): pass\n");
write_file(&root, "b.py", "def bar(): pass\n");
let mut state = WatchState::new();
let (analysis, cs, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert_eq!(cs.files_added, 2);
assert_eq!(analysis.graph.file_count(), 2);
std::fs::remove_file(&a).unwrap();
let (analysis2, cs2, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert_eq!(cs2.files_removed, 1);
assert_eq!(analysis2.graph.file_count(), 1);
assert_eq!(state.cache.extractions.len(), 1);
}
#[test]
fn detects_added_file() {
let root = temp_dir("add");
write_file(&root, "a.py", "def foo(): pass\n");
let mut state = WatchState::new();
let (_, cs, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert_eq!(cs.files_added, 1);
write_file(&root, "b.py", "def bar(): pass\n");
let (analysis2, cs2, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert_eq!(cs2.files_added, 1);
assert_eq!(analysis2.graph.file_count(), 2);
}
#[test]
fn symbol_ids_no_collision_on_re_extract() {
let root = temp_dir("idcol");
write_file(&root, "a.py", "def one(): pass\n");
write_file(&root, "b.py", "def two(): pass\n");
let mut state = WatchState::new();
let (_, _, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
write_file(&root, "c.py", "def three(): pass\n");
let (analysis, _, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
let mut ids: Vec<u32> = analysis
.graph
.symbols()
.map(|(id, _)| id.to_raw())
.collect();
ids.sort();
let expected: Vec<u32> = (1..=3).collect();
assert_eq!(ids, expected, "symbol IDs must be unique and contiguous");
}
#[cfg(feature = "dataflow")]
#[test]
fn data_node_ids_no_collision_on_re_extract() {
let root = temp_dir("data_idcol");
write_file(&root, "a.py", "x = 1\ny = x + 1\n");
write_file(&root, "b.py", "z = 2\n");
let mut state = WatchState::new();
let (_, _, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
write_file(&root, "b.py", "z = 2\nw = z + 3\n");
let (_, _, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
let mut ids: Vec<u32> = state
.cache
.extractions
.values()
.flat_map(|ext| ext.data_nodes.iter().map(|d| d.id.to_raw()))
.collect();
let original_len = ids.len();
ids.sort();
ids.dedup();
assert_eq!(
ids.len(),
original_len,
"data node IDs must be unique across re-extractions"
);
}
#[test]
fn diff_counts_are_exact() {
let root = temp_dir("diffcounts");
let a = write_file(&root, "a.py", "def a(): pass\n");
let _b = write_file(&root, "b.py", "def b(): pass\n");
let mut state = WatchState::new();
let (_, cs1, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert_eq!(cs1.files_added, 2);
assert_eq!(cs1.files_removed, 0);
std::fs::remove_file(&a).unwrap();
let (_, cs2, _) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert_eq!(cs2.files_removed, 1);
assert_eq!(cs2.files_added, 0);
assert_eq!(cs2.files_modified, 0);
assert_eq!(cs2.files_unchanged, 1);
}
#[test]
fn unreadable_file_emits_diagnostic() {
let root = temp_dir("unread_diag");
let a = write_file(&root, "a.py", "def a(): pass\n");
let mut state = WatchState::new();
let (_, _, diags1) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert!(diags1.is_empty());
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut perms = std::fs::metadata(&a).unwrap().permissions();
perms.set_mode(0o000);
let _ = std::fs::set_permissions(&a, perms);
}
let (_, _, diags2) = incremental_reanalyze(&root, None, &mut state).unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut perms = std::fs::metadata(&a).unwrap().permissions();
perms.set_mode(0o644);
let _ = std::fs::set_permissions(&a, perms);
}
#[cfg(unix)]
assert!(!diags2.is_empty(), "Unreadable file must emit diagnostic");
}
#[test]
fn empty_project_handled() {
let root = temp_dir("empty");
let mut state = WatchState::new();
let (analysis, cs, diags) = incremental_reanalyze(&root, None, &mut state).unwrap();
assert!(diags.is_empty());
assert_eq!(cs.files_added, 0);
assert_eq!(analysis.graph.node_count(), 0);
}
#[test]
fn overlay_overrides_disk_text() {
let root = temp_dir("overlay");
write_file(&root, "a.py", "def disk(): pass\n");
let mut state = WatchState::new();
let first = overlay(&root, "a.py", "def from_buffer(): pass\n");
let (extractions, cs, _) =
reanalyze_extractions(&root, None, std::slice::from_ref(&first), &mut state).unwrap();
assert_eq!(cs.files_added, 1);
let names = symbol_names(&extractions);
assert!(names.contains(&"from_buffer".to_string()));
assert!(!names.contains(&"disk".to_string()));
let (_, cs2, _) = reanalyze_extractions(&root, None, &[first], &mut state).unwrap();
assert_eq!(cs2.files_unchanged, 1);
let (extractions3, cs3, _) = reanalyze_extractions(&root, None, &[], &mut state).unwrap();
assert_eq!(cs3.files_modified, 1);
assert!(symbol_names(&extractions3).contains(&"disk".to_string()));
}
#[test]
fn overlay_adds_file_not_on_disk() {
let root = temp_dir("overlay_new");
let mut state = WatchState::new();
let pending = overlay(&root, "new.py", "def fresh(): pass\n");
let (extractions, cs, _) =
reanalyze_extractions(&root, None, &[pending], &mut state).unwrap();
assert_eq!(cs.files_added, 1);
assert_eq!(extractions.len(), 1);
assert!(symbol_names(&extractions).contains(&"fresh".to_string()));
}
#[test]
fn overlay_outside_root_is_ignored() {
let root = temp_dir("overlay_outside");
let mut state = WatchState::new();
let outside = Overlay {
uri: "file:///elsewhere.py".to_string(),
path: PathBuf::from("/definitely/outside/elsewhere.py"),
text: "def ignored(): pass\n".to_string(),
version: 1,
lang: LangId::Python,
};
let (extractions, cs, _) =
reanalyze_extractions(&root, None, &[outside], &mut state).unwrap();
assert!(extractions.is_empty());
assert_eq!(cs.files_added, 0);
}
#[test]
fn snapshot_id_allocation_is_monotonic() {
let mut state = WatchState::new();
let s1 = state.next_snapshot_id();
let s2 = state.next_snapshot_id();
assert_eq!(s1.to_raw(), 1);
assert_eq!(s2.to_raw(), 2);
}
}