1use std::{
2 collections::{BTreeSet, VecDeque},
3 path::{Path, PathBuf},
4 sync::{Arc, OnceLock},
5 time::{Duration, Instant, SystemTime, UNIX_EPOCH},
6};
7
8use tokio::sync::{Semaphore, oneshot};
9
10use crate::{
11 api::{
12 ApiError, ApiMetadata, FileContentQueryRequest, FileContentQueryResponse,
13 FileIndexFreshnessState, FileIndexRequest, FileIndexResponse, FileQueryRequest,
14 FileQueryResponse, RequestContext,
15 },
16 domain::{FreshnessPolicy, GraphVersion},
17 storage::{
18 FileContentSearchRequest, FileIndexEntry, FileIndexRoot, FileIndexRootUpdate,
19 FileIndexScanSummary, FileSearchRequest, StorageError,
20 },
21};
22
23use crate::application::{FileIndexRootConfig, service::RelayKnowledgeService};
24
25#[path = "file_content_budget.rs"]
26mod file_content_budget;
27#[path = "file_content_extract.rs"]
28mod file_content_extract;
29#[path = "file_content_read.rs"]
30mod file_content_read;
31
32use file_content_extract::{FileContentEntryResult, file_content_entry, text_content_extension};
33use file_content_read::MAX_CONTENT_INDEX_BYTES;
34
35use super::file_freshness::{FileFreshnessContext, file_freshness_diagnostics};
36
37pub const DEFAULT_FILE_QUERY_LIMIT: usize = 20;
38const MAX_FILE_QUERY_LIMIT: usize = 500;
39const MAX_CONCURRENT_FILE_SCANS: usize = 4;
40const MAX_CONTENT_SCAN_BYTES: usize = 64 * 1024 * 1024;
41static FILE_SCAN_LIMITER: OnceLock<Arc<Semaphore>> = OnceLock::new();
42
43#[derive(Clone)]
44struct ScanBudget {
45 max_depth: usize,
46 max_file_bytes: u64,
47 max_files_per_root: usize,
48 excludes: Vec<String>,
49}
50
51impl RelayKnowledgeService {
52 pub async fn index_files(
54 &self,
55 request: FileIndexRequest,
56 context: RequestContext,
57 ) -> Result<FileIndexResponse, ApiError> {
58 let configured_scan = request.roots.is_empty();
59 let roots = self
60 .file_index_roots_from_request(request)
61 .map_err(ApiError::invalid_argument)?;
62 let active_roots = roots
63 .iter()
64 .map(file_index_root_from_config)
65 .collect::<Vec<_>>();
66 let store = self.storage.get().await.map_err(storage_api_error)?;
67 let now_ms = current_time_millis();
68 let updates = scan_roots(
69 roots,
70 ScanBudget {
71 max_depth: self.runtime.file_index.max_depth,
72 max_file_bytes: self.runtime.file_index.max_file_bytes,
73 max_files_per_root: self.runtime.file_index.max_files_per_root,
74 excludes: self.runtime.file_index.excludes.clone(),
75 },
76 now_ms,
77 self.runtime.file_index.scan_timeout,
78 )
79 .await
80 .map_err(storage_api_error)?;
81 let mut summary = FileIndexScanSummary::default();
82 for update in updates {
83 let status = store
84 .replace_file_index_root(update)
85 .await
86 .map_err(storage_api_error)?;
87 summary.root_count = summary.root_count.saturating_add(1);
88 summary.indexed_file_count = summary
89 .indexed_file_count
90 .saturating_add(status.indexed_file_count);
91 summary.missing_file_count = summary
92 .missing_file_count
93 .saturating_add(status.missing_file_count);
94 summary.indexed_content_count = summary
95 .indexed_content_count
96 .saturating_add(status.indexed_content_count);
97 summary.skipped_content_count = summary
98 .skipped_content_count
99 .saturating_add(status.skipped_content_count);
100 summary.unchanged_content_count = summary
101 .unchanged_content_count
102 .saturating_add(status.unchanged_content_count);
103 summary.stale_content_cursor_count = summary
104 .stale_content_cursor_count
105 .saturating_add(status.stale_content_cursor_count);
106 summary.scan_error_count = summary
107 .scan_error_count
108 .saturating_add(status.scan_error_count);
109 summary.content_read_error_count = summary
110 .content_read_error_count
111 .saturating_add(status.content_read_error_count);
112 if status.truncated {
113 summary.truncated_root_count = summary.truncated_root_count.saturating_add(1);
114 }
115 summary.roots.push(status);
116 }
117 if configured_scan {
118 let diagnostics = store
119 .mark_file_index_roots_unconfigured(active_roots, now_ms)
120 .await
121 .map_err(storage_api_error)?;
122 summary = summary_from_diagnostics(diagnostics);
123 }
124
125 Ok(FileIndexResponse {
126 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
127 summary,
128 })
129 }
130
131 pub async fn index_configured_files_once(&self) -> Result<FileIndexResponse, ApiError> {
133 if self.runtime.file_index.roots.is_empty() {
134 let store = self.storage.get().await.map_err(storage_api_error)?;
135 let diagnostics = store
136 .mark_file_index_roots_unconfigured(Vec::new(), current_time_millis())
137 .await
138 .map_err(storage_api_error)?;
139 return Ok(FileIndexResponse {
140 metadata: ApiMetadata::graph_only(
141 &RequestContext::for_interface(crate::api::InterfaceKind::Cli),
142 GraphVersion::ZERO,
143 ),
144 summary: summary_from_diagnostics(diagnostics),
145 });
146 }
147
148 self.index_files(
149 FileIndexRequest {
150 source_scope: None,
151 roots: Vec::new(),
152 },
153 RequestContext::for_interface(crate::api::InterfaceKind::Cli),
154 )
155 .await
156 }
157
158 pub async fn query_files(
160 &self,
161 request: FileQueryRequest,
162 context: RequestContext,
163 ) -> Result<FileQueryResponse, ApiError> {
164 let query = required_query(request.query).map_err(ApiError::invalid_argument)?;
165 let limit = bounded_limit(request.limit).map_err(ApiError::invalid_argument)?;
166 let store = self.storage.get().await.map_err(storage_api_error)?;
167 let started = Instant::now();
168 let source_scope =
169 normalize_optional_text(request.source_scope).map_err(ApiError::invalid_argument)?;
170 let root_id =
171 normalize_optional_text(request.root_id).map_err(ApiError::invalid_argument)?;
172 let configured_roots = self
173 .runtime
174 .file_index
175 .roots
176 .iter()
177 .map(file_index_root_from_config)
178 .collect::<Vec<_>>();
179 let diagnostics = store
180 .file_index_diagnostics()
181 .await
182 .map_err(storage_api_error)?;
183 if request.freshness_policy == FreshnessPolicy::GraphOnly {
184 let degraded_reason = "graph_only freshness policy selected".to_owned();
185 let freshness = file_freshness_diagnostics(FileFreshnessContext {
186 file_index_enabled: self.runtime.file_index.enabled,
187 configured_roots: &configured_roots,
188 diagnostics: &diagnostics,
189 freshness_policy: request.freshness_policy,
190 source_scope: source_scope.clone(),
191 root_id: root_id.clone(),
192 graph_version: GraphVersion::ZERO.get(),
193 query_degraded_reason: Some(degraded_reason.clone()),
194 returned_paths: &[],
195 content_required: false,
196 });
197 return Ok(FileQueryResponse {
198 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
199 query,
200 source_scope,
201 root_id,
202 freshness,
203 results: Vec::new(),
204 truncated: false,
205 duration_ms: elapsed_ms(started),
206 degraded_reason: Some(degraded_reason),
207 });
208 }
209 let freshness = file_freshness_diagnostics(FileFreshnessContext {
210 file_index_enabled: self.runtime.file_index.enabled,
211 configured_roots: &configured_roots,
212 diagnostics: &diagnostics,
213 freshness_policy: request.freshness_policy,
214 source_scope: source_scope.clone(),
215 root_id: root_id.clone(),
216 graph_version: GraphVersion::ZERO.get(),
217 query_degraded_reason: None,
218 returned_paths: &[],
219 content_required: false,
220 });
221 if request.freshness_policy == FreshnessPolicy::WaitUntilFresh
222 && freshness.state != FileIndexFreshnessState::Fresh
223 {
224 return Err(ApiError::invalid_argument(format!(
225 "file index is {}; run files index before querying with wait_until_fresh",
226 file_freshness_state_label(freshness.state)
227 )));
228 }
229 let results = match store
230 .search_files(FileSearchRequest {
231 query: query.clone(),
232 source_scope: source_scope.clone(),
233 root_id: root_id.clone(),
234 limit: limit.saturating_add(1),
235 timeout_ms: query_timeout_ms(self.runtime.file_index.query_timeout),
236 })
237 .await
238 {
239 Ok(results) => results,
240 Err(error) if storage_error_timed_out(&error) => {
241 let degraded_reason = "file query timed out".to_owned();
242 let freshness = file_freshness_diagnostics(FileFreshnessContext {
243 file_index_enabled: self.runtime.file_index.enabled,
244 configured_roots: &configured_roots,
245 diagnostics: &diagnostics,
246 freshness_policy: request.freshness_policy,
247 source_scope: source_scope.clone(),
248 root_id: root_id.clone(),
249 graph_version: GraphVersion::ZERO.get(),
250 query_degraded_reason: Some(degraded_reason.clone()),
251 returned_paths: &[],
252 content_required: false,
253 });
254 return Ok(FileQueryResponse {
255 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
256 query,
257 source_scope,
258 root_id,
259 freshness,
260 results: Vec::new(),
261 truncated: false,
262 duration_ms: elapsed_ms(started),
263 degraded_reason: Some(degraded_reason),
264 });
265 }
266 Err(error) => return Err(storage_api_error(error)),
267 };
268 let mut results = results;
269 let truncated = results.len() > limit;
270 results.truncate(limit);
271 let result_paths = results
272 .iter()
273 .map(|hit| hit.path.clone())
274 .collect::<Vec<_>>();
275 let freshness = file_freshness_diagnostics(FileFreshnessContext {
276 file_index_enabled: self.runtime.file_index.enabled,
277 configured_roots: &configured_roots,
278 diagnostics: &diagnostics,
279 freshness_policy: request.freshness_policy,
280 source_scope: source_scope.clone(),
281 root_id: root_id.clone(),
282 graph_version: GraphVersion::ZERO.get(),
283 query_degraded_reason: None,
284 returned_paths: &result_paths,
285 content_required: false,
286 });
287
288 Ok(FileQueryResponse {
289 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
290 query,
291 source_scope,
292 root_id,
293 freshness,
294 results,
295 truncated,
296 duration_ms: elapsed_ms(started),
297 degraded_reason: None,
298 })
299 }
300
301 pub async fn query_file_content(
303 &self,
304 request: FileContentQueryRequest,
305 context: RequestContext,
306 ) -> Result<FileContentQueryResponse, ApiError> {
307 let query = required_query(request.query).map_err(ApiError::invalid_argument)?;
308 let limit = bounded_limit(request.limit).map_err(ApiError::invalid_argument)?;
309 let store = self.storage.get().await.map_err(storage_api_error)?;
310 let started = Instant::now();
311 let source_scope =
312 normalize_optional_text(request.source_scope).map_err(ApiError::invalid_argument)?;
313 let root_id =
314 normalize_optional_text(request.root_id).map_err(ApiError::invalid_argument)?;
315 let configured_roots = self
316 .runtime
317 .file_index
318 .roots
319 .iter()
320 .map(file_index_root_from_config)
321 .collect::<Vec<_>>();
322 let diagnostics = store
323 .file_index_diagnostics()
324 .await
325 .map_err(storage_api_error)?;
326 if request.freshness_policy == FreshnessPolicy::GraphOnly {
327 let degraded_reason = "graph_only freshness policy selected".to_owned();
328 let freshness = file_freshness_diagnostics(FileFreshnessContext {
329 file_index_enabled: self.runtime.file_index.enabled,
330 configured_roots: &configured_roots,
331 diagnostics: &diagnostics,
332 freshness_policy: request.freshness_policy,
333 source_scope: source_scope.clone(),
334 root_id: root_id.clone(),
335 graph_version: GraphVersion::ZERO.get(),
336 query_degraded_reason: Some(degraded_reason.clone()),
337 returned_paths: &[],
338 content_required: true,
339 });
340 return Ok(FileContentQueryResponse {
341 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
342 query,
343 source_scope,
344 root_id,
345 freshness,
346 results: Vec::new(),
347 truncated: false,
348 duration_ms: elapsed_ms(started),
349 degraded_reason: Some(degraded_reason),
350 });
351 }
352 let freshness = file_freshness_diagnostics(FileFreshnessContext {
353 file_index_enabled: self.runtime.file_index.enabled,
354 configured_roots: &configured_roots,
355 diagnostics: &diagnostics,
356 freshness_policy: request.freshness_policy,
357 source_scope: source_scope.clone(),
358 root_id: root_id.clone(),
359 graph_version: GraphVersion::ZERO.get(),
360 query_degraded_reason: None,
361 returned_paths: &[],
362 content_required: true,
363 });
364 if request.freshness_policy == FreshnessPolicy::WaitUntilFresh
365 && freshness.state != FileIndexFreshnessState::Fresh
366 {
367 return Err(ApiError::invalid_argument(format!(
368 "file content index is {}; run files index before querying with wait_until_fresh",
369 file_freshness_state_label(freshness.state)
370 )));
371 }
372 let results = match store
373 .search_file_content(FileContentSearchRequest {
374 query: query.clone(),
375 source_scope: source_scope.clone(),
376 root_id: root_id.clone(),
377 authorized_roots: configured_roots.clone(),
378 limit: limit.saturating_add(1),
379 timeout_ms: query_timeout_ms(self.runtime.file_index.query_timeout),
380 })
381 .await
382 {
383 Ok(results) => results,
384 Err(error) if storage_error_timed_out(&error) => {
385 let degraded_reason = "file content query timed out".to_owned();
386 let freshness = file_freshness_diagnostics(FileFreshnessContext {
387 file_index_enabled: self.runtime.file_index.enabled,
388 configured_roots: &configured_roots,
389 diagnostics: &diagnostics,
390 freshness_policy: request.freshness_policy,
391 source_scope: source_scope.clone(),
392 root_id: root_id.clone(),
393 graph_version: GraphVersion::ZERO.get(),
394 query_degraded_reason: Some(degraded_reason.clone()),
395 returned_paths: &[],
396 content_required: true,
397 });
398 return Ok(FileContentQueryResponse {
399 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
400 query,
401 source_scope,
402 root_id,
403 freshness,
404 results: Vec::new(),
405 truncated: false,
406 duration_ms: elapsed_ms(started),
407 degraded_reason: Some(degraded_reason),
408 });
409 }
410 Err(error) => return Err(storage_api_error(error)),
411 };
412 let mut results = results;
413 let truncated = results.len() > limit;
414 results.truncate(limit);
415 let result_paths = results
416 .iter()
417 .map(|hit| hit.path.clone())
418 .collect::<Vec<_>>();
419 let freshness = file_freshness_diagnostics(FileFreshnessContext {
420 file_index_enabled: self.runtime.file_index.enabled,
421 configured_roots: &configured_roots,
422 diagnostics: &diagnostics,
423 freshness_policy: request.freshness_policy,
424 source_scope: source_scope.clone(),
425 root_id: root_id.clone(),
426 graph_version: GraphVersion::ZERO.get(),
427 query_degraded_reason: None,
428 returned_paths: &result_paths,
429 content_required: true,
430 });
431
432 Ok(FileContentQueryResponse {
433 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
434 query,
435 source_scope,
436 root_id,
437 freshness,
438 results,
439 truncated,
440 duration_ms: elapsed_ms(started),
441 degraded_reason: None,
442 })
443 }
444
445 fn file_index_roots_from_request(
446 &self,
447 request: FileIndexRequest,
448 ) -> Result<Vec<FileIndexRootConfig>, String> {
449 if request.roots.is_empty() {
450 if self.runtime.file_index.roots.is_empty() {
451 return Err("no file index roots are configured".to_owned());
452 }
453 return Ok(self.runtime.file_index.roots.clone());
454 }
455
456 let scope_id = normalize_optional_text(request.source_scope)?
457 .unwrap_or_else(|| "local-files".to_owned());
458 if self.runtime.file_index.roots.is_empty() {
459 return Err(
460 "file index roots must be configured before explicit roots can be scanned"
461 .to_owned(),
462 );
463 }
464 let mut roots = request
465 .roots
466 .into_iter()
467 .map(|root| {
468 let root = root.trim();
469 if root.is_empty() {
470 Err("file index root must not be empty".to_owned())
471 } else {
472 let root_path = PathBuf::from(root);
473 if !root_path.is_absolute() {
474 return Err("file index root must be an absolute path".to_owned());
475 }
476 let requested = FileIndexRootConfig::new(&scope_id, root_path);
477 self.runtime
478 .file_index
479 .roots
480 .iter()
481 .find(|authorized| {
482 authorized.scope_id == requested.scope_id
483 && authorized.root_id == requested.root_id
484 })
485 .cloned()
486 .ok_or_else(|| {
487 format!(
488 "file index root '{root}' is not configured for scope '{scope_id}'"
489 )
490 })
491 }
492 })
493 .collect::<Result<Vec<_>, _>>()?;
494 roots.sort_by(|left, right| {
495 left.scope_id
496 .cmp(&right.scope_id)
497 .then(left.root_id.cmp(&right.root_id))
498 });
499 roots.dedup_by(|left, right| {
500 left.scope_id == right.scope_id && left.root_id == right.root_id
501 });
502
503 Ok(roots)
504 }
505}
506
507async fn scan_roots(
508 roots: Vec<FileIndexRootConfig>,
509 budget: ScanBudget,
510 now_ms: u64,
511 scan_timeout: Duration,
512) -> Result<Vec<FileIndexRootUpdate>, StorageError> {
513 let mut updates = Vec::with_capacity(roots.len());
514 for root in roots {
515 updates.push(scan_root_with_timeout(root, budget.clone(), now_ms, scan_timeout).await?);
516 }
517
518 Ok(updates)
519}
520
521async fn scan_root_with_timeout(
522 root: FileIndexRootConfig,
523 budget: ScanBudget,
524 now_ms: u64,
525 scan_timeout: Duration,
526) -> Result<FileIndexRootUpdate, StorageError> {
527 if scan_timeout.is_zero() {
528 return Ok(timed_out_file_index_root_update(root, now_ms));
529 }
530 let permit = match file_scan_limiter().try_acquire_owned() {
531 Ok(permit) => permit,
532 Err(_) => return Ok(scan_worker_busy_file_index_root_update(root, now_ms)),
533 };
534 let timeout_root = root.clone();
535 let (sender, receiver) = oneshot::channel();
536 std::thread::Builder::new()
537 .name("relay-file-index-scan".to_owned())
538 .spawn(move || {
539 let _permit = permit;
540 let _ = sender.send(scan_root(root, &budget, now_ms));
541 })?;
542
543 match tokio::time::timeout(scan_timeout, receiver).await {
544 Ok(Ok(result)) => result,
545 Ok(Err(_)) => Err(StorageError::InvalidInput(
546 "file index scan worker stopped before reporting".to_owned(),
547 )),
548 Err(_) => Ok(timed_out_file_index_root_update(timeout_root, now_ms)),
549 }
550}
551
552fn file_scan_limiter() -> Arc<Semaphore> {
553 Arc::clone(
554 FILE_SCAN_LIMITER.get_or_init(|| Arc::new(Semaphore::new(MAX_CONCURRENT_FILE_SCANS))),
555 )
556}
557
558fn scan_worker_busy_file_index_root_update(
559 root: FileIndexRootConfig,
560 now_ms: u64,
561) -> FileIndexRootUpdate {
562 FileIndexRootUpdate {
563 root: storage_root(root.scope_id, root.root_id, &root.root_path),
564 entries: Vec::new(),
565 processed_content_paths: BTreeSet::new(),
566 content_entries: Vec::new(),
567 scan_error_count: 1,
568 truncated: true,
569 content_truncated: false,
570 content_read_error_count: 0,
571 last_error: Some("file index scan worker is still busy".to_owned()),
572 now_ms,
573 }
574}
575
576fn timed_out_file_index_root_update(root: FileIndexRootConfig, now_ms: u64) -> FileIndexRootUpdate {
577 FileIndexRootUpdate {
578 root: storage_root(root.scope_id, root.root_id, &root.root_path),
579 entries: Vec::new(),
580 processed_content_paths: BTreeSet::new(),
581 content_entries: Vec::new(),
582 scan_error_count: 1,
583 truncated: true,
584 content_truncated: false,
585 content_read_error_count: 0,
586 last_error: Some("file index scan timed out".to_owned()),
587 now_ms,
588 }
589}
590
591fn scan_root(
592 root: FileIndexRootConfig,
593 budget: &ScanBudget,
594 now_ms: u64,
595) -> Result<FileIndexRootUpdate, StorageError> {
596 let root_path = root.root_path;
597 let mut entries = Vec::new();
598 let mut processed_content_paths = BTreeSet::new();
599 let mut content_entries = Vec::new();
600 let mut content_scan_bytes = 0usize;
601 let mut scan_error_count = 0usize;
602 let mut truncated = false;
603 let mut content_truncated = false;
604 let mut content_read_error_count = 0usize;
605 let mut last_error = None;
606 let canonical_root = match std::fs::canonicalize(&root_path) {
607 Ok(path) => path,
608 Err(error) => {
609 return Ok(FileIndexRootUpdate {
610 root: storage_root(root.scope_id, root.root_id, &root_path),
611 entries,
612 processed_content_paths,
613 content_entries,
614 scan_error_count: 1,
615 truncated: false,
616 content_truncated: false,
617 content_read_error_count: 0,
618 last_error: Some(error.to_string()),
619 now_ms,
620 });
621 }
622 };
623 let mut pending = VecDeque::from([(canonical_root.clone(), 0usize)]);
624
625 while let Some((directory, depth)) = pending.pop_front() {
626 if entries.len() >= budget.max_files_per_root {
627 truncated = true;
628 break;
629 }
630 if depth > budget.max_depth {
631 truncated = true;
632 continue;
633 }
634 let read_dir = match std::fs::read_dir(&directory) {
635 Ok(read_dir) => read_dir,
636 Err(error) => {
637 scan_error_count = scan_error_count.saturating_add(1);
638 last_error = Some(error.to_string());
639 continue;
640 }
641 };
642 for child in read_dir {
643 if entries.len() >= budget.max_files_per_root {
644 truncated = true;
645 pending.clear();
646 break;
647 }
648 let child = match child {
649 Ok(child) => child,
650 Err(error) => {
651 scan_error_count = scan_error_count.saturating_add(1);
652 last_error = Some(error.to_string());
653 continue;
654 }
655 };
656 let path = child.path();
657 if excluded(&path, &budget.excludes) {
658 continue;
659 }
660 let file_type = match child.file_type() {
661 Ok(file_type) => file_type,
662 Err(error) => {
663 scan_error_count = scan_error_count.saturating_add(1);
664 last_error = Some(error.to_string());
665 continue;
666 }
667 };
668 if file_type.is_symlink() {
669 continue;
670 }
671 if file_type.is_dir() {
672 pending.push_back((path, depth.saturating_add(1)));
673 continue;
674 }
675 let metadata = match child.metadata() {
676 Ok(metadata) => metadata,
677 Err(error) => {
678 scan_error_count = scan_error_count.saturating_add(1);
679 last_error = Some(error.to_string());
680 continue;
681 }
682 };
683 if file_type.is_file() && metadata.len() <= budget.max_file_bytes {
684 let entry = file_entry(
685 &root.scope_id,
686 &root.root_id,
687 &canonical_root,
688 &path,
689 &metadata,
690 );
691 if text_content_extension(entry.extension.as_deref()) {
692 if metadata.len() > MAX_CONTENT_INDEX_BYTES {
693 processed_content_paths.insert(entry.path.clone());
694 } else if content_scan_bytes < MAX_CONTENT_SCAN_BYTES {
695 if file_content_budget::reserve_content_read_with_budget(
696 &mut content_scan_bytes,
697 metadata.len(),
698 MAX_CONTENT_SCAN_BYTES,
699 ) {
700 content_truncated = true;
701 } else {
702 match file_content_entry(
703 &entry,
704 &metadata,
705 &canonical_root,
706 now_ms,
707 GraphVersion::ZERO.get(),
708 ) {
709 FileContentEntryResult::Indexed(content_entry) => {
710 processed_content_paths.insert(entry.path.clone());
711 content_entries.push(*content_entry);
712 }
713 FileContentEntryResult::Skipped => {
714 processed_content_paths.insert(entry.path.clone());
715 }
716 FileContentEntryResult::ReadFailed => {
717 content_read_error_count =
718 content_read_error_count.saturating_add(1);
719 last_error.get_or_insert_with(|| {
720 "file content read failed".to_owned()
721 });
722 }
723 }
724 }
725 } else if content_scan_bytes >= MAX_CONTENT_SCAN_BYTES {
726 content_truncated = true;
727 }
728 }
729 entries.push(entry);
730 }
731 }
732 }
733
734 Ok(FileIndexRootUpdate {
735 root: storage_root(root.scope_id, root.root_id, &canonical_root),
736 entries,
737 processed_content_paths,
738 content_entries,
739 scan_error_count,
740 truncated,
741 content_truncated,
742 content_read_error_count,
743 last_error,
744 now_ms,
745 })
746}
747
748fn file_index_root_from_config(root: &FileIndexRootConfig) -> FileIndexRoot {
749 FileIndexRoot {
750 scope_id: root.scope_id.clone(),
751 root_id: root.root_id.clone(),
752 root_path: root.root_path.to_string_lossy().to_string(),
753 }
754}
755
756fn summary_from_diagnostics(
757 diagnostics: crate::storage::FileIndexDiagnostics,
758) -> FileIndexScanSummary {
759 FileIndexScanSummary {
760 root_count: diagnostics.root_count,
761 indexed_file_count: diagnostics.indexed_file_count,
762 missing_file_count: diagnostics.missing_file_count,
763 indexed_content_count: diagnostics.indexed_content_count,
764 skipped_content_count: diagnostics.skipped_content_count,
765 unchanged_content_count: diagnostics.unchanged_content_count,
766 stale_content_cursor_count: diagnostics.stale_content_cursor_count,
767 scan_error_count: diagnostics.scan_error_count,
768 content_read_error_count: diagnostics.content_read_error_count,
769 truncated_root_count: diagnostics.truncated_root_count,
770 roots: diagnostics.roots,
771 }
772}
773
774fn file_entry(
775 scope_id: &str,
776 root_id: &str,
777 root: &Path,
778 path: &Path,
779 metadata: &std::fs::Metadata,
780) -> FileIndexEntry {
781 let relative_path = path.strip_prefix(root).unwrap_or(path);
782 let file_name = path
783 .file_name()
784 .map(|value| value.to_string_lossy().to_string())
785 .unwrap_or_default();
786 let extension = path
787 .extension()
788 .map(|value| value.to_string_lossy().to_ascii_lowercase());
789 let parent_dir = path
790 .parent()
791 .map(|value| value.to_string_lossy().to_string())
792 .unwrap_or_default();
793 let modified_at_ms = metadata
794 .modified()
795 .ok()
796 .and_then(system_time_millis)
797 .unwrap_or_default();
798
799 FileIndexEntry {
800 scope_id: scope_id.to_owned(),
801 root_id: root_id.to_owned(),
802 path: path.to_string_lossy().to_string(),
803 relative_path: relative_path.to_string_lossy().to_string(),
804 file_name,
805 extension,
806 parent_dir,
807 size_bytes: metadata.len(),
808 modified_at_ms,
809 fingerprint: format!("{}:{modified_at_ms}", metadata.len()),
810 }
811}
812
813fn storage_root(scope_id: String, root_id: String, root_path: &Path) -> FileIndexRoot {
814 FileIndexRoot {
815 scope_id,
816 root_id,
817 root_path: root_path.to_string_lossy().to_string(),
818 }
819}
820
821fn excluded(path: &Path, configured: &[String]) -> bool {
822 let Some(name) = path.file_name().map(|value| value.to_string_lossy()) else {
823 return false;
824 };
825 if name.starts_with('.') {
826 return true;
827 }
828 let lower = name.to_ascii_lowercase();
829 matches!(
830 lower.as_str(),
831 "target" | "node_modules" | ".git" | "__pycache__" | "tmp" | "temp" | "cache"
832 ) || configured
833 .iter()
834 .any(|pattern| lower.contains(&pattern.to_ascii_lowercase()))
835}
836
837fn required_query(query: String) -> Result<String, String> {
838 let query = query.trim().to_owned();
839 if query.is_empty() {
840 Err("file query must not be empty".to_owned())
841 } else {
842 Ok(query)
843 }
844}
845
846fn bounded_limit(limit: usize) -> Result<usize, String> {
847 match limit {
848 0 => Err("file query limit must be greater than zero".to_owned()),
849 value if value > MAX_FILE_QUERY_LIMIT => Err(format!(
850 "file query limit must not exceed {MAX_FILE_QUERY_LIMIT}"
851 )),
852 value => Ok(value),
853 }
854}
855
856fn normalize_optional_text(value: Option<String>) -> Result<Option<String>, String> {
857 value
858 .map(|value| {
859 let value = value.trim().to_owned();
860 if value.is_empty() {
861 Err("optional file query filter must not be empty".to_owned())
862 } else {
863 Ok(value)
864 }
865 })
866 .transpose()
867}
868
869fn current_time_millis() -> u64 {
870 SystemTime::now()
871 .duration_since(UNIX_EPOCH)
872 .map_or(0, |duration| {
873 u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
874 })
875}
876
877fn system_time_millis(time: SystemTime) -> Option<u64> {
878 time.duration_since(UNIX_EPOCH)
879 .ok()
880 .map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
881}
882
883fn elapsed_ms(started: Instant) -> u64 {
884 u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
885}
886
887fn query_timeout_ms(timeout: std::time::Duration) -> u64 {
888 u64::try_from(timeout.as_millis()).unwrap_or(u64::MAX)
889}
890
891fn storage_error_timed_out(error: &StorageError) -> bool {
892 matches!(
893 error,
894 StorageError::InvalidInput(message)
895 if message.contains("file query timed out")
896 || message.contains("file content query timed out")
897 )
898}
899
900fn file_freshness_state_label(state: FileIndexFreshnessState) -> &'static str {
901 match state {
902 FileIndexFreshnessState::Fresh => "fresh",
903 FileIndexFreshnessState::Pending => "pending",
904 FileIndexFreshnessState::Paused => "paused",
905 FileIndexFreshnessState::Stale => "stale",
906 FileIndexFreshnessState::Degraded => "degraded",
907 FileIndexFreshnessState::Overflow => "overflow",
908 }
909}
910
911fn storage_api_error(error: StorageError) -> ApiError {
912 ApiError::storage_unavailable(error.to_string())
913}
914
915#[cfg(test)]
916#[path = "file_index_tests.rs"]
917mod tests;