file_engine/analysis/
builder.rs

1use std::path::PathBuf;
2use std::time::{Instant, SystemTime};
3
4use tokio_util::sync::CancellationToken;
5
6use crate::error::Result;
7
8use super::error_strategy::AnalysisErrorStrategy;
9use super::filter::AnalysisFilter;
10use super::handle::AnalysisHandle;
11use super::progress::AnalysisProgressReporter;
12use super::report::{AnalysisReport, DEFAULT_MAX_REPORTED_ERRORS};
13use super::util::default_concurrency;
14use super::walk::{walk, WalkParams};
15
16#[cfg(feature = "checksum")]
17use super::hash::detect_duplicates;
18#[cfg(feature = "checksum")]
19use super::report::DEFAULT_MAX_REPORTED_DUPLICATE_GROUPS;
20
21/// Default cap on `largest_files` — see `AnalyzeBuilder::top_n_largest`.
22pub const DEFAULT_TOP_N_LARGEST: usize = 10;
23
24pub struct AnalyzeBuilder {
25    root: PathBuf,
26    filter: AnalysisFilter,
27    error_strategy: AnalysisErrorStrategy,
28    max_depth: Option<usize>,
29    follow_symlinks: bool,
30    top_n_largest: usize,
31    detect_mime_types: bool,
32    max_reported_errors: usize,
33    walk_concurrency: Option<usize>,
34    #[cfg(feature = "checksum")]
35    detect_duplicates: bool,
36    #[cfg(feature = "checksum")]
37    hash_concurrency: Option<usize>,
38    #[cfg(feature = "checksum")]
39    max_reported_duplicates: usize,
40}
41
42impl AnalyzeBuilder {
43    pub(crate) fn new(root: impl Into<PathBuf>) -> Self {
44        Self {
45            root: root.into(),
46            filter: AnalysisFilter::default(),
47            error_strategy: AnalysisErrorStrategy::default(),
48            max_depth: None,
49            follow_symlinks: false,
50            top_n_largest: DEFAULT_TOP_N_LARGEST,
51            detect_mime_types: false,
52            max_reported_errors: DEFAULT_MAX_REPORTED_ERRORS,
53            walk_concurrency: None,
54            #[cfg(feature = "checksum")]
55            detect_duplicates: false,
56            #[cfg(feature = "checksum")]
57            hash_concurrency: None,
58            #[cfg(feature = "checksum")]
59            max_reported_duplicates: DEFAULT_MAX_REPORTED_DUPLICATE_GROUPS,
60        }
61    }
62
63    /// Only files with one of these extensions (case-insensitive,
64    /// without the leading dot) are matched. Unset matches any
65    /// extension, including files with none.
66    pub fn extensions(mut self, exts: impl IntoIterator<Item = impl Into<String>>) -> Self {
67        self.filter.extensions = Some(exts.into_iter().map(Into::into).collect());
68        self
69    }
70
71    /// Glob patterns (matched against the path relative to the analyzed
72    /// root) that prune traversal entirely — an excluded directory is
73    /// never descended into, not merely omitted from the report.
74    pub fn exclude_globs(mut self, patterns: impl IntoIterator<Item = impl Into<String>>) -> Self {
75        self.filter.exclude_patterns = patterns.into_iter().map(Into::into).collect();
76        self
77    }
78
79    pub fn min_size(mut self, bytes: u64) -> Self {
80        self.filter.min_size = Some(bytes);
81        self
82    }
83
84    pub fn max_size(mut self, bytes: u64) -> Self {
85        self.filter.max_size = Some(bytes);
86        self
87    }
88
89    /// Only files modified at or after `t` are matched. A file with no
90    /// readable modified time never matches once this is set.
91    pub fn modified_after(mut self, t: SystemTime) -> Self {
92        self.filter.modified_after = Some(t);
93        self
94    }
95
96    /// Only files modified at or before `t` are matched. A file with no
97    /// readable modified time never matches once this is set.
98    pub fn modified_before(mut self, t: SystemTime) -> Self {
99        self.filter.modified_before = Some(t);
100        self
101    }
102
103    /// Bounds how far the walk descends: the analyzed root is depth 0,
104    /// its immediate children depth 1, and so on — passed straight
105    /// through to `jwalk`'s own `max_depth`, which prunes traversal past
106    /// the bound rather than filtering after the fact.
107    pub fn max_depth(mut self, depth: usize) -> Self {
108        self.max_depth = Some(depth);
109        self
110    }
111
112    /// Off by default. When enabled, `jwalk`'s own loop detection
113    /// surfaces a symlink cycle as a per-entry error, handled like any
114    /// other error via `.on_error()`.
115    pub fn follow_symlinks(mut self, follow: bool) -> Self {
116        self.follow_symlinks = follow;
117        self
118    }
119
120    /// How many worker threads `jwalk` uses to read directories and stat
121    /// entries concurrently while walking. Defaults to
122    /// `available_parallelism()`, matching `hash_concurrency`.
123    ///
124    /// Traversal is still reported (and aggregated into `AnalysisReport`)
125    /// in a single stream on the calling task — this only parallelizes
126    /// the underlying directory reads and `stat` calls, which is where
127    /// wall time actually goes on large trees or slow/network
128    /// filesystems.
129    pub fn walk_concurrency(mut self, n: usize) -> Self {
130        self.walk_concurrency = Some(n);
131        self
132    }
133
134    /// How many of the largest matched files to keep in
135    /// `AnalysisReport::largest_files`. `0` disables the tracking
136    /// entirely.
137    pub fn top_n_largest(mut self, n: usize) -> Self {
138        self.top_n_largest = n;
139        self
140    }
141
142    /// Off by default — sniffing every matched file's header via `infer`
143    /// is an extra read per file, on top of the metadata `walkdir`
144    /// already reads for every entry.
145    pub fn detect_mime_types(mut self, enable: bool) -> Self {
146        self.detect_mime_types = enable;
147        self
148    }
149
150    pub fn on_error(mut self, strategy: AnalysisErrorStrategy) -> Self {
151        self.error_strategy = strategy;
152        self
153    }
154
155    /// Caps `AnalysisReport::errors`; `AnalysisReport::errors_total`
156    /// stays uncapped regardless of this setting.
157    pub fn max_reported_errors(mut self, n: usize) -> Self {
158        self.max_reported_errors = n;
159        self
160    }
161
162    /// Off by default — hashing every matched file (even the pre-filtered
163    /// size-collision candidates) is real I/O on top of the walk itself.
164    #[cfg(feature = "checksum")]
165    pub fn detect_duplicates(mut self, enable: bool) -> Self {
166        self.detect_duplicates = enable;
167        self
168    }
169
170    /// Defaults to `available_parallelism()`, matching
171    /// `CopyBuilder::batch_concurrency`.
172    #[cfg(feature = "checksum")]
173    pub fn hash_concurrency(mut self, n: usize) -> Self {
174        self.hash_concurrency = Some(n);
175        self
176    }
177
178    /// Caps `AnalysisReport::duplicates`; `duplicate_groups_total` and
179    /// `duplicate_bytes_wasted` stay uncapped regardless of this setting.
180    #[cfg(feature = "checksum")]
181    pub fn max_reported_duplicates(mut self, n: usize) -> Self {
182        self.max_reported_duplicates = n;
183        self
184    }
185
186    pub fn start(self) -> Result<AnalysisHandle> {
187        let cancel = CancellationToken::new();
188        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
189        let reporter = AnalysisProgressReporter::new(tx);
190        let cancel_for_task = cancel.clone();
191
192        #[cfg(feature = "checksum")]
193        let detect_duplicates_enabled = self.detect_duplicates;
194
195        let params = WalkParams {
196            root: self.root,
197            filter: self.filter,
198            error_strategy: self.error_strategy,
199            max_depth: self.max_depth,
200            follow_symlinks: self.follow_symlinks,
201            top_n_largest: self.top_n_largest,
202            detect_mime_types: self.detect_mime_types,
203            #[cfg(feature = "checksum")]
204            collect_duplicate_candidates: detect_duplicates_enabled,
205            max_reported_errors: self.max_reported_errors,
206            walk_concurrency: self.walk_concurrency.unwrap_or_else(default_concurrency),
207        };
208
209        #[cfg(feature = "checksum")]
210        let error_strategy = self.error_strategy;
211        #[cfg(feature = "checksum")]
212        let hash_concurrency = self.hash_concurrency;
213        #[cfg(feature = "checksum")]
214        let max_reported_duplicates = self.max_reported_duplicates;
215        #[cfg(feature = "checksum")]
216        let max_reported_errors = self.max_reported_errors;
217
218        let join_handle = tokio::spawn(async move {
219            let started = Instant::now();
220            let outcome = walk(params, cancel_for_task.clone(), reporter.clone()).await?;
221
222            #[cfg(feature = "checksum")]
223            let (duplicates, duplicate_groups_total, duplicate_bytes_wasted, errors, errors_total) = {
224                let mut errors = outcome.errors;
225                let mut errors_total = outcome.errors_total;
226                if detect_duplicates_enabled {
227                    let hash_outcome = detect_duplicates(
228                        outcome.duplicate_candidates,
229                        hash_concurrency,
230                        max_reported_duplicates,
231                        max_reported_errors.saturating_sub(errors.len()),
232                        error_strategy,
233                        &cancel_for_task,
234                        &reporter,
235                    )
236                    .await?;
237                    errors.extend(hash_outcome.errors);
238                    errors_total += hash_outcome.errors_total;
239                    (
240                        hash_outcome.groups,
241                        hash_outcome.groups_total,
242                        hash_outcome.bytes_wasted,
243                        errors,
244                        errors_total,
245                    )
246                } else {
247                    (Vec::new(), 0, 0, errors, errors_total)
248                }
249            };
250            #[cfg(not(feature = "checksum"))]
251            let (errors, errors_total) = (outcome.errors, outcome.errors_total);
252
253            Ok(AnalysisReport {
254                file_count: outcome.file_count,
255                dir_count: outcome.dir_count,
256                total_size: outcome.total_size,
257                largest_files: outcome.largest_files,
258                by_extension: outcome.by_extension,
259                by_mime: outcome.by_mime,
260                age_buckets: outcome.age_buckets,
261                errors,
262                errors_total,
263                #[cfg(feature = "checksum")]
264                duplicates,
265                #[cfg(feature = "checksum")]
266                duplicate_groups_total,
267                #[cfg(feature = "checksum")]
268                duplicate_bytes_wasted,
269                duration: started.elapsed(),
270            })
271        });
272
273        Ok(AnalysisHandle::new(join_handle, rx, cancel))
274    }
275}
file_engine/analysis/builder.rs

file_engine/analysis/
builder.rs
