relay_knowledge/application/
file_index.rs

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

relay_knowledge/application/
file_index.rs
