1use std::{
2 path::PathBuf,
3 time::{Instant, SystemTime, UNIX_EPOCH},
4};
5
6use crate::{
7 api::{
8 ApiError, ApiMetadata, FileContentQueryRequest, FileContentQueryResponse,
9 FileIndexFreshnessState, FileIndexRequest, FileIndexResponse, FileQueryRequest,
10 FileQueryResponse, RequestContext,
11 },
12 domain::{FreshnessPolicy, GraphVersion},
13 storage::{FileContentSearchRequest, FileIndexScanSummary, FileSearchRequest, StorageError},
14};
15
16use crate::application::{FileIndexRootConfig, service::RelayKnowledgeService};
17
18mod content;
19mod scanner;
20#[cfg(test)]
21#[path = "test_support_tests.rs"]
22mod test_support;
23
24use super::file_freshness::{FileFreshnessContext, file_freshness_diagnostics};
25use scanner::{ScanBudget, file_index_root_from_config, scan_roots, summary_from_diagnostics};
26
27pub const DEFAULT_FILE_QUERY_LIMIT: usize = 20;
28const MAX_FILE_QUERY_LIMIT: usize = 500;
29
30impl RelayKnowledgeService {
31 pub async fn index_files(
33 &self,
34 request: FileIndexRequest,
35 context: RequestContext,
36 ) -> Result<FileIndexResponse, ApiError> {
37 let configured_scan = request.roots.is_empty();
38 let roots = self
39 .file_index_roots_from_request(request)
40 .map_err(ApiError::invalid_argument)?;
41 let active_roots = roots
42 .iter()
43 .map(file_index_root_from_config)
44 .collect::<Vec<_>>();
45 let store = self.storage.get().await.map_err(storage_api_error)?;
46 let now_ms = current_time_millis();
47 let updates = scan_roots(
48 roots,
49 ScanBudget {
50 max_depth: self.runtime.file_index.max_depth,
51 max_file_bytes: self.runtime.file_index.max_file_bytes,
52 max_files_per_root: self.runtime.file_index.max_files_per_root,
53 excludes: self.runtime.file_index.excludes.clone(),
54 },
55 now_ms,
56 self.runtime.file_index.scan_timeout,
57 )
58 .await
59 .map_err(storage_api_error)?;
60 let mut summary = FileIndexScanSummary::default();
61 for update in updates {
62 let status = store
63 .replace_file_index_root(update)
64 .await
65 .map_err(storage_api_error)?;
66 summary.root_count = summary.root_count.saturating_add(1);
67 summary.indexed_file_count = summary
68 .indexed_file_count
69 .saturating_add(status.indexed_file_count);
70 summary.missing_file_count = summary
71 .missing_file_count
72 .saturating_add(status.missing_file_count);
73 summary.indexed_content_count = summary
74 .indexed_content_count
75 .saturating_add(status.indexed_content_count);
76 summary.skipped_content_count = summary
77 .skipped_content_count
78 .saturating_add(status.skipped_content_count);
79 summary.unchanged_content_count = summary
80 .unchanged_content_count
81 .saturating_add(status.unchanged_content_count);
82 summary.stale_content_cursor_count = summary
83 .stale_content_cursor_count
84 .saturating_add(status.stale_content_cursor_count);
85 summary.scan_error_count = summary
86 .scan_error_count
87 .saturating_add(status.scan_error_count);
88 summary.content_read_error_count = summary
89 .content_read_error_count
90 .saturating_add(status.content_read_error_count);
91 if status.truncated {
92 summary.truncated_root_count = summary.truncated_root_count.saturating_add(1);
93 }
94 summary.roots.push(status);
95 }
96 if configured_scan {
97 let diagnostics = store
98 .mark_file_index_roots_unconfigured(active_roots, now_ms)
99 .await
100 .map_err(storage_api_error)?;
101 summary = summary_from_diagnostics(diagnostics);
102 }
103
104 Ok(FileIndexResponse {
105 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
106 summary,
107 })
108 }
109
110 pub async fn index_configured_files_once(&self) -> Result<FileIndexResponse, ApiError> {
112 if self.runtime.file_index.roots.is_empty() {
113 let store = self.storage.get().await.map_err(storage_api_error)?;
114 let diagnostics = store
115 .mark_file_index_roots_unconfigured(Vec::new(), current_time_millis())
116 .await
117 .map_err(storage_api_error)?;
118 return Ok(FileIndexResponse {
119 metadata: ApiMetadata::graph_only(
120 &RequestContext::for_interface(crate::api::InterfaceKind::Cli),
121 GraphVersion::ZERO,
122 ),
123 summary: summary_from_diagnostics(diagnostics),
124 });
125 }
126
127 self.index_files(
128 FileIndexRequest {
129 source_scope: None,
130 roots: Vec::new(),
131 },
132 RequestContext::for_interface(crate::api::InterfaceKind::Cli),
133 )
134 .await
135 }
136
137 pub async fn query_files(
139 &self,
140 request: FileQueryRequest,
141 context: RequestContext,
142 ) -> Result<FileQueryResponse, ApiError> {
143 let query = required_query(request.query).map_err(ApiError::invalid_argument)?;
144 let limit = bounded_limit(request.limit).map_err(ApiError::invalid_argument)?;
145 let store = self.storage.get().await.map_err(storage_api_error)?;
146 let started = Instant::now();
147 let source_scope =
148 normalize_optional_text(request.source_scope).map_err(ApiError::invalid_argument)?;
149 let root_id =
150 normalize_optional_text(request.root_id).map_err(ApiError::invalid_argument)?;
151 let configured_roots = self
152 .runtime
153 .file_index
154 .roots
155 .iter()
156 .map(file_index_root_from_config)
157 .collect::<Vec<_>>();
158 let diagnostics = store
159 .file_index_diagnostics()
160 .await
161 .map_err(storage_api_error)?;
162 if request.freshness_policy == FreshnessPolicy::GraphOnly {
163 let degraded_reason = "graph_only freshness policy selected".to_owned();
164 let freshness = file_freshness_diagnostics(FileFreshnessContext {
165 file_index_enabled: self.runtime.file_index.enabled,
166 configured_roots: &configured_roots,
167 diagnostics: &diagnostics,
168 freshness_policy: request.freshness_policy,
169 source_scope: source_scope.clone(),
170 root_id: root_id.clone(),
171 graph_version: GraphVersion::ZERO.get(),
172 query_degraded_reason: Some(degraded_reason.clone()),
173 returned_paths: &[],
174 content_required: false,
175 });
176 return Ok(FileQueryResponse {
177 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
178 query,
179 source_scope,
180 root_id,
181 freshness,
182 results: Vec::new(),
183 truncated: false,
184 duration_ms: elapsed_ms(started),
185 degraded_reason: Some(degraded_reason),
186 });
187 }
188 let freshness = file_freshness_diagnostics(FileFreshnessContext {
189 file_index_enabled: self.runtime.file_index.enabled,
190 configured_roots: &configured_roots,
191 diagnostics: &diagnostics,
192 freshness_policy: request.freshness_policy,
193 source_scope: source_scope.clone(),
194 root_id: root_id.clone(),
195 graph_version: GraphVersion::ZERO.get(),
196 query_degraded_reason: None,
197 returned_paths: &[],
198 content_required: false,
199 });
200 if request.freshness_policy == FreshnessPolicy::WaitUntilFresh
201 && freshness.state != FileIndexFreshnessState::Fresh
202 {
203 return Err(ApiError::invalid_argument(format!(
204 "file index is {}; run files index before querying with wait_until_fresh",
205 file_freshness_state_label(freshness.state)
206 )));
207 }
208 let results = match store
209 .search_files(FileSearchRequest {
210 query: query.clone(),
211 source_scope: source_scope.clone(),
212 root_id: root_id.clone(),
213 limit: limit.saturating_add(1),
214 timeout_ms: query_timeout_ms(self.runtime.file_index.query_timeout),
215 })
216 .await
217 {
218 Ok(results) => results,
219 Err(error) if storage_error_timed_out(&error) => {
220 let degraded_reason = "file query timed out".to_owned();
221 let freshness = file_freshness_diagnostics(FileFreshnessContext {
222 file_index_enabled: self.runtime.file_index.enabled,
223 configured_roots: &configured_roots,
224 diagnostics: &diagnostics,
225 freshness_policy: request.freshness_policy,
226 source_scope: source_scope.clone(),
227 root_id: root_id.clone(),
228 graph_version: GraphVersion::ZERO.get(),
229 query_degraded_reason: Some(degraded_reason.clone()),
230 returned_paths: &[],
231 content_required: false,
232 });
233 return Ok(FileQueryResponse {
234 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
235 query,
236 source_scope,
237 root_id,
238 freshness,
239 results: Vec::new(),
240 truncated: false,
241 duration_ms: elapsed_ms(started),
242 degraded_reason: Some(degraded_reason),
243 });
244 }
245 Err(error) => return Err(storage_api_error(error)),
246 };
247 let mut results = results;
248 let truncated = results.len() > limit;
249 results.truncate(limit);
250 let result_paths = results
251 .iter()
252 .map(|hit| hit.path.clone())
253 .collect::<Vec<_>>();
254 let freshness = file_freshness_diagnostics(FileFreshnessContext {
255 file_index_enabled: self.runtime.file_index.enabled,
256 configured_roots: &configured_roots,
257 diagnostics: &diagnostics,
258 freshness_policy: request.freshness_policy,
259 source_scope: source_scope.clone(),
260 root_id: root_id.clone(),
261 graph_version: GraphVersion::ZERO.get(),
262 query_degraded_reason: None,
263 returned_paths: &result_paths,
264 content_required: false,
265 });
266
267 Ok(FileQueryResponse {
268 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
269 query,
270 source_scope,
271 root_id,
272 freshness,
273 results,
274 truncated,
275 duration_ms: elapsed_ms(started),
276 degraded_reason: None,
277 })
278 }
279
280 pub async fn query_file_content(
282 &self,
283 request: FileContentQueryRequest,
284 context: RequestContext,
285 ) -> Result<FileContentQueryResponse, ApiError> {
286 let query = required_query(request.query).map_err(ApiError::invalid_argument)?;
287 let limit = bounded_limit(request.limit).map_err(ApiError::invalid_argument)?;
288 let store = self.storage.get().await.map_err(storage_api_error)?;
289 let started = Instant::now();
290 let source_scope =
291 normalize_optional_text(request.source_scope).map_err(ApiError::invalid_argument)?;
292 let root_id =
293 normalize_optional_text(request.root_id).map_err(ApiError::invalid_argument)?;
294 let configured_roots = self
295 .runtime
296 .file_index
297 .roots
298 .iter()
299 .map(file_index_root_from_config)
300 .collect::<Vec<_>>();
301 let diagnostics = store
302 .file_index_diagnostics()
303 .await
304 .map_err(storage_api_error)?;
305 if request.freshness_policy == FreshnessPolicy::GraphOnly {
306 let degraded_reason = "graph_only freshness policy selected".to_owned();
307 let freshness = file_freshness_diagnostics(FileFreshnessContext {
308 file_index_enabled: self.runtime.file_index.enabled,
309 configured_roots: &configured_roots,
310 diagnostics: &diagnostics,
311 freshness_policy: request.freshness_policy,
312 source_scope: source_scope.clone(),
313 root_id: root_id.clone(),
314 graph_version: GraphVersion::ZERO.get(),
315 query_degraded_reason: Some(degraded_reason.clone()),
316 returned_paths: &[],
317 content_required: true,
318 });
319 return Ok(FileContentQueryResponse {
320 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
321 query,
322 source_scope,
323 root_id,
324 freshness,
325 results: Vec::new(),
326 truncated: false,
327 duration_ms: elapsed_ms(started),
328 degraded_reason: Some(degraded_reason),
329 });
330 }
331 let freshness = file_freshness_diagnostics(FileFreshnessContext {
332 file_index_enabled: self.runtime.file_index.enabled,
333 configured_roots: &configured_roots,
334 diagnostics: &diagnostics,
335 freshness_policy: request.freshness_policy,
336 source_scope: source_scope.clone(),
337 root_id: root_id.clone(),
338 graph_version: GraphVersion::ZERO.get(),
339 query_degraded_reason: None,
340 returned_paths: &[],
341 content_required: true,
342 });
343 if request.freshness_policy == FreshnessPolicy::WaitUntilFresh
344 && freshness.state != FileIndexFreshnessState::Fresh
345 {
346 return Err(ApiError::invalid_argument(format!(
347 "file content index is {}; run files index before querying with wait_until_fresh",
348 file_freshness_state_label(freshness.state)
349 )));
350 }
351 let results = match store
352 .search_file_content(FileContentSearchRequest {
353 query: query.clone(),
354 source_scope: source_scope.clone(),
355 root_id: root_id.clone(),
356 authorized_roots: configured_roots.clone(),
357 limit: limit.saturating_add(1),
358 timeout_ms: query_timeout_ms(self.runtime.file_index.query_timeout),
359 })
360 .await
361 {
362 Ok(results) => results,
363 Err(error) if storage_error_timed_out(&error) => {
364 let degraded_reason = "file content query timed out".to_owned();
365 let freshness = file_freshness_diagnostics(FileFreshnessContext {
366 file_index_enabled: self.runtime.file_index.enabled,
367 configured_roots: &configured_roots,
368 diagnostics: &diagnostics,
369 freshness_policy: request.freshness_policy,
370 source_scope: source_scope.clone(),
371 root_id: root_id.clone(),
372 graph_version: GraphVersion::ZERO.get(),
373 query_degraded_reason: Some(degraded_reason.clone()),
374 returned_paths: &[],
375 content_required: true,
376 });
377 return Ok(FileContentQueryResponse {
378 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
379 query,
380 source_scope,
381 root_id,
382 freshness,
383 results: Vec::new(),
384 truncated: false,
385 duration_ms: elapsed_ms(started),
386 degraded_reason: Some(degraded_reason),
387 });
388 }
389 Err(error) => return Err(storage_api_error(error)),
390 };
391 let mut results = results;
392 let truncated = results.len() > limit;
393 results.truncate(limit);
394 let result_paths = results
395 .iter()
396 .map(|hit| hit.path.clone())
397 .collect::<Vec<_>>();
398 let freshness = file_freshness_diagnostics(FileFreshnessContext {
399 file_index_enabled: self.runtime.file_index.enabled,
400 configured_roots: &configured_roots,
401 diagnostics: &diagnostics,
402 freshness_policy: request.freshness_policy,
403 source_scope: source_scope.clone(),
404 root_id: root_id.clone(),
405 graph_version: GraphVersion::ZERO.get(),
406 query_degraded_reason: None,
407 returned_paths: &result_paths,
408 content_required: true,
409 });
410
411 Ok(FileContentQueryResponse {
412 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
413 query,
414 source_scope,
415 root_id,
416 freshness,
417 results,
418 truncated,
419 duration_ms: elapsed_ms(started),
420 degraded_reason: None,
421 })
422 }
423
424 fn file_index_roots_from_request(
425 &self,
426 request: FileIndexRequest,
427 ) -> Result<Vec<FileIndexRootConfig>, String> {
428 if request.roots.is_empty() {
429 if self.runtime.file_index.roots.is_empty() {
430 return Err("no file index roots are configured".to_owned());
431 }
432 return Ok(self.runtime.file_index.roots.clone());
433 }
434
435 let scope_id = normalize_optional_text(request.source_scope)?
436 .unwrap_or_else(|| "local-files".to_owned());
437 if self.runtime.file_index.roots.is_empty() {
438 return Err(
439 "file index roots must be configured before explicit roots can be scanned"
440 .to_owned(),
441 );
442 }
443 let mut roots = request
444 .roots
445 .into_iter()
446 .map(|root| {
447 let root = root.trim();
448 if root.is_empty() {
449 Err("file index root must not be empty".to_owned())
450 } else {
451 let root_path = PathBuf::from(root);
452 if !root_path.is_absolute() {
453 return Err("file index root must be an absolute path".to_owned());
454 }
455 let requested = FileIndexRootConfig::new(&scope_id, root_path);
456 self.runtime
457 .file_index
458 .roots
459 .iter()
460 .find(|authorized| {
461 authorized.scope_id == requested.scope_id
462 && authorized.root_id == requested.root_id
463 })
464 .cloned()
465 .ok_or_else(|| {
466 format!(
467 "file index root '{root}' is not configured for scope '{scope_id}'"
468 )
469 })
470 }
471 })
472 .collect::<Result<Vec<_>, _>>()?;
473 roots.sort_by(|left, right| {
474 left.scope_id
475 .cmp(&right.scope_id)
476 .then(left.root_id.cmp(&right.root_id))
477 });
478 roots.dedup_by(|left, right| {
479 left.scope_id == right.scope_id && left.root_id == right.root_id
480 });
481
482 Ok(roots)
483 }
484}
485fn required_query(query: String) -> Result<String, String> {
486 let query = query.trim().to_owned();
487 if query.is_empty() {
488 Err("file query must not be empty".to_owned())
489 } else {
490 Ok(query)
491 }
492}
493
494fn bounded_limit(limit: usize) -> Result<usize, String> {
495 match limit {
496 0 => Err("file query limit must be greater than zero".to_owned()),
497 value if value > MAX_FILE_QUERY_LIMIT => Err(format!(
498 "file query limit must not exceed {MAX_FILE_QUERY_LIMIT}"
499 )),
500 value => Ok(value),
501 }
502}
503
504fn normalize_optional_text(value: Option<String>) -> Result<Option<String>, String> {
505 value
506 .map(|value| {
507 let value = value.trim().to_owned();
508 if value.is_empty() {
509 Err("optional file query filter must not be empty".to_owned())
510 } else {
511 Ok(value)
512 }
513 })
514 .transpose()
515}
516
517fn current_time_millis() -> u64 {
518 SystemTime::now()
519 .duration_since(UNIX_EPOCH)
520 .map_or(0, |duration| {
521 u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
522 })
523}
524
525fn elapsed_ms(started: Instant) -> u64 {
526 u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
527}
528
529fn query_timeout_ms(timeout: std::time::Duration) -> u64 {
530 u64::try_from(timeout.as_millis()).unwrap_or(u64::MAX)
531}
532
533fn storage_error_timed_out(error: &StorageError) -> bool {
534 matches!(
535 error,
536 StorageError::InvalidInput(message)
537 if message.contains("file query timed out")
538 || message.contains("file content query timed out")
539 )
540}
541
542fn file_freshness_state_label(state: FileIndexFreshnessState) -> &'static str {
543 match state {
544 FileIndexFreshnessState::Fresh => "fresh",
545 FileIndexFreshnessState::Pending => "pending",
546 FileIndexFreshnessState::Paused => "paused",
547 FileIndexFreshnessState::Stale => "stale",
548 FileIndexFreshnessState::Degraded => "degraded",
549 FileIndexFreshnessState::Overflow => "overflow",
550 }
551}
552
553fn storage_api_error(error: StorageError) -> ApiError {
554 ApiError::storage_unavailable(error.to_string())
555}
556
557#[cfg(test)]
558#[path = "workflow_tests.rs"]
559mod workflow_tests;