weavatrix_rust/
analyzer.rs

1mod imports;
2mod mounts;
3mod references;
4mod state;
5mod support;
6
7use crate::Result;
8use crate::error::Error;
9use crate::language::LanguageRegistry;
10use crate::snapshot::Snapshot;
11use state::{AnalysisState, ParsedSource, parse_source};
12use std::collections::BTreeSet;
13use std::path::Path;
14use std::sync::atomic::{AtomicUsize, Ordering};
15use std::sync::{Arc, Mutex};
16use support::{canonical_repository, capabilities};
17use weavatrix_scan::{
18    ContentDiscoveryMode, ContentFileStatus, ContentVisitControl, ContentVisitEvent, ScanOptions,
19    ScanReport, Scanner,
20};
21
22/// Runs `fetch` for every index across all available cores and returns the
23/// results in index order, so downstream integration stays deterministic.
24fn parse_parallel<F>(count: usize, fetch: F) -> Result<Vec<ParsedSource>>
25where
26    F: Fn(usize) -> Result<ParsedSource> + Sync,
27{
28    let workers = std::thread::available_parallelism()
29        .map_or(1, usize::from)
30        .min(count)
31        .max(1);
32    if workers <= 1 {
33        return (0..count).map(fetch).collect();
34    }
35    let cursor = AtomicUsize::new(0);
36    let results = Mutex::new(Vec::with_capacity(count));
37    std::thread::scope(|scope| {
38        for _ in 0..workers {
39            scope.spawn(|| {
40                loop {
41                    let index = cursor.fetch_add(1, Ordering::Relaxed);
42                    if index >= count {
43                        break;
44                    }
45                    let result = fetch(index);
46                    let mut guard = results
47                        .lock()
48                        .unwrap_or_else(std::sync::PoisonError::into_inner);
49                    guard.push((index, result));
50                }
51            });
52        }
53    });
54    let mut items = results
55        .into_inner()
56        .unwrap_or_else(std::sync::PoisonError::into_inner);
57    items.sort_unstable_by_key(|(index, _)| *index);
58    items.into_iter().map(|(_, result)| result).collect()
59}
60
61#[derive(Debug, Clone)]
62pub struct AnalyzerConfig {
63    pub max_file_bytes: u64,
64}
65
66impl Default for AnalyzerConfig {
67    fn default() -> Self {
68        Self {
69            max_file_bytes: 1_500_000,
70        }
71    }
72}
73
74pub struct Analyzer {
75    config: AnalyzerConfig,
76    languages: LanguageRegistry,
77}
78
79#[derive(Debug, Clone)]
80pub struct SourceInput {
81    pub path: String,
82    pub bytes: Vec<u8>,
83    pub content_hash: Option<String>,
84}
85
86impl Default for Analyzer {
87    fn default() -> Self {
88        Self::new(AnalyzerConfig::default())
89    }
90}
91
92impl Analyzer {
93    #[must_use]
94    pub fn new(config: AnalyzerConfig) -> Self {
95        Self {
96            config,
97            languages: LanguageRegistry::default(),
98        }
99    }
100
101    #[must_use]
102    pub fn supports_path(&self, path: &str) -> bool {
103        Path::new(path)
104            .extension()
105            .and_then(|value| value.to_str())
106            .map(str::to_ascii_lowercase)
107            .is_some_and(|extension| self.languages.adapter_for_extension(&extension).is_some())
108    }
109
110    #[must_use]
111    pub const fn max_file_bytes(&self) -> u64 {
112        self.config.max_file_bytes
113    }
114
115    /// Analyzes a repository into a deterministic, evidence-carrying snapshot.
116    ///
117    /// # Errors
118    ///
119    /// Returns an error when the repository cannot be read, an adapter cannot
120    /// initialize, or normalized facts violate graph integrity.
121    pub fn analyze(&self, repository: impl AsRef<Path>) -> Result<Snapshot> {
122        self.analyze_with_report(repository)
123            .map(|(snapshot, _)| snapshot)
124    }
125
126    pub(crate) fn analyze_with_report(
127        &self,
128        repository: impl AsRef<Path>,
129    ) -> Result<(Snapshot, ScanReport)> {
130        let repository = canonical_repository(repository.as_ref())?;
131        let timing = std::env::var_os("WEAVATRIX_PHASE_TIMING").is_some();
132        let started = std::time::Instant::now();
133        let parsed = Arc::new(Mutex::new(Vec::<(u64, Result<ParsedSource>)>::new()));
134        let sink = Arc::clone(&parsed);
135        let visit = Scanner::new(&repository)
136            .options(self.scan_options())
137            .visit_content_manifest(move |_| {
138                let sink = Arc::clone(&sink);
139                let registry = LanguageRegistry::default();
140                let mut bytes = Vec::new();
141                move |event| {
142                    match event {
143                        ContentVisitEvent::FileStart { file, .. } => {
144                            bytes.clear();
145                            if let Ok(required) = usize::try_from(file.bytes)
146                                && required > bytes.capacity()
147                            {
148                                bytes.reserve(required - bytes.capacity());
149                            }
150                        }
151                        ContentVisitEvent::Chunk { bytes: chunk, .. } => {
152                            bytes.extend_from_slice(chunk);
153                        }
154                        ContentVisitEvent::FileEnd {
155                            file,
156                            status: ContentFileStatus::Selected,
157                            content_hash,
158                            ..
159                        } => {
160                            let result =
161                                parse_source(file.relative, &bytes, content_hash, &registry);
162                            sink.lock()
163                                .unwrap_or_else(std::sync::PoisonError::into_inner)
164                                .push((file.sequence, result));
165                        }
166                        ContentVisitEvent::FileEnd { .. } => bytes.clear(),
167                    }
168                    ContentVisitControl::Continue
169                }
170            })?;
171        let scan = visit.into_scan_report();
172        let mut parsed = {
173            let mut guard = parsed
174                .lock()
175                .unwrap_or_else(std::sync::PoisonError::into_inner);
176            std::mem::take(&mut *guard)
177        };
178        parsed.sort_unstable_by_key(|(sequence, _)| *sequence);
179        let mut parsed = parsed
180            .into_iter()
181            .map(|(_, result)| result)
182            .collect::<Result<Vec<_>>>()?;
183        mounts::apply(&mut parsed);
184        let parsed_at = started.elapsed();
185        let (snapshot, integrated_at, resolved_at) =
186            self.integrate_snapshot(&repository, &scan, parsed, &started)?;
187        if timing {
188            eprintln!(
189                "phase-timing one-pass-parse={:.1}ms integrate={:.1}ms resolve={:.1}ms snapshot={:.1}ms",
190                parsed_at.as_secs_f64() * 1e3,
191                integrated_at.saturating_sub(parsed_at).as_secs_f64() * 1e3,
192                resolved_at.saturating_sub(integrated_at).as_secs_f64() * 1e3,
193                started.elapsed().saturating_sub(resolved_at).as_secs_f64() * 1e3,
194            );
195        }
196        Ok((snapshot, scan))
197    }
198
199    pub(crate) fn scan(
200        &self,
201        repository: &Path,
202        previous: Option<&ScanReport>,
203    ) -> Result<ScanReport> {
204        let options = self.scan_options();
205        let scanner = Scanner::new(repository).options(options);
206        Ok(match previous {
207            Some(previous) => scanner.scan_incremental(previous)?,
208            None => scanner.scan()?,
209        })
210    }
211
212    pub(crate) fn analyze_report(&self, repository: &Path, scan: &ScanReport) -> Result<Snapshot> {
213        let timing = std::env::var_os("WEAVATRIX_PHASE_TIMING").is_some();
214        let started = std::time::Instant::now();
215        let mut parsed = parse_parallel(scan.files.len(), |index| {
216            let file = &scan.files[index];
217            let bytes = std::fs::read(&file.absolute)
218                .map_err(|source| Error::io(&file.absolute, source))?;
219            parse_source(
220                &file.relative,
221                &bytes,
222                file.content_hash.as_deref(),
223                &self.languages,
224            )
225        })?;
226        mounts::apply(&mut parsed);
227        let parsed_at = started.elapsed();
228        let (snapshot, integrated_at, resolved_at) =
229            self.integrate_snapshot(repository, scan, parsed, &started)?;
230        if timing {
231            eprintln!(
232                "phase-timing parse={:.1}ms integrate={:.1}ms resolve={:.1}ms snapshot={:.1}ms",
233                parsed_at.as_secs_f64() * 1e3,
234                integrated_at.saturating_sub(parsed_at).as_secs_f64() * 1e3,
235                resolved_at.saturating_sub(integrated_at).as_secs_f64() * 1e3,
236                started.elapsed().saturating_sub(resolved_at).as_secs_f64() * 1e3,
237            );
238        }
239        Ok(snapshot)
240    }
241
242    fn integrate_snapshot(
243        &self,
244        repository: &Path,
245        scan: &ScanReport,
246        parsed: Vec<ParsedSource>,
247        started: &std::time::Instant,
248    ) -> Result<(Snapshot, std::time::Duration, std::time::Duration)> {
249        let (node_hint, edge_hint) = AnalysisState::expected(&parsed);
250        let mut state = AnalysisState::with_capacity(repository, node_hint, edge_hint)?;
251        state.add_scan_warnings(scan.warnings.clone());
252        for item in parsed {
253            state.integrate(item)?;
254        }
255        let integrated_at = started.elapsed();
256        state.resolve_references()?;
257        let resolved_at = started.elapsed();
258        let snapshot = state.into_snapshot(
259            repository,
260            scan.revision.clone(),
261            capabilities(&self.languages),
262        )?;
263        Ok((snapshot, integrated_at, resolved_at))
264    }
265
266    fn scan_options(&self) -> ScanOptions {
267        let extensions = self
268            .languages
269            .extensions()
270            .map(str::to_owned)
271            .collect::<BTreeSet<_>>();
272        let mut options = ScanOptions::default().with_extensions(extensions);
273        options.max_file_bytes = self.config.max_file_bytes;
274        options.content_discovery = ContentDiscoveryMode::BufferedParallel;
275        options
276    }
277
278    /// Analyzes an immutable set of source blobs without materializing a tree.
279    ///
280    /// This is the bridge used by the Git module for revision-aware graph
281    /// comparisons. The repository path only supplies stable repository
282    /// identity; every analyzed byte comes from `sources`.
283    ///
284    /// # Errors
285    ///
286    /// Returns parser or graph validation failures.
287    pub fn analyze_sources(
288        &self,
289        repository: impl AsRef<Path>,
290        revision: impl Into<String>,
291        sources: impl IntoIterator<Item = SourceInput>,
292    ) -> Result<Snapshot> {
293        let repository = canonical_repository(repository.as_ref())?;
294        let sources = sources
295            .into_iter()
296            .filter(|source| {
297                u64::try_from(source.bytes.len()).unwrap_or(u64::MAX) <= self.config.max_file_bytes
298            })
299            .collect::<Vec<_>>();
300        let mut parsed = parse_parallel(sources.len(), |index| {
301            let source = &sources[index];
302            parse_source(
303                &source.path,
304                &source.bytes,
305                source.content_hash.as_deref(),
306                &self.languages,
307            )
308        })?;
309        mounts::apply(&mut parsed);
310        let (node_hint, edge_hint) = AnalysisState::expected(&parsed);
311        let mut state = AnalysisState::with_capacity(&repository, node_hint, edge_hint)?;
312        for item in parsed {
313            state.integrate(item)?;
314        }
315        state.resolve_references()?;
316        state.into_snapshot(&repository, revision.into(), capabilities(&self.languages))
317    }
318
319    /// Analyzes a repository and serializes the snapshot as JSON.
320    ///
321    /// # Errors
322    ///
323    /// Returns any analysis error or a JSON serialization error.
324    pub fn analyze_json(&self, repository: impl AsRef<Path>, pretty: bool) -> Result<String> {
325        let snapshot = self.analyze(repository)?;
326        if pretty {
327            Ok(blazingly_json::to_string_pretty(&snapshot)?)
328        } else {
329            Ok(blazingly_json::to_string(&snapshot)?)
330        }
331    }
332
333    /// Analyzes a repository and serializes a JS Weavatrix-compatible graph.
334    ///
335    /// # Errors
336    ///
337    /// Returns any analysis error or a JSON serialization error.
338    pub fn analyze_legacy_json(
339        &self,
340        repository: impl AsRef<Path>,
341        pretty: bool,
342    ) -> Result<String> {
343        Ok(self.analyze(repository)?.legacy_json(pretty)?)
344    }
345}
weavatrix_rust/analyzer.rs

weavatrix_rust/
analyzer.rs
