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