relay_knowledge/application/service/
watcher.rs

1use std::{path::PathBuf, sync::Arc};
2
3use crate::{
4    api::ApiError,
5    domain::CodeRepositoryStatus,
6    watcher::{FileWatcher, WatchedRepository, WatcherHandle},
7};
8
9use super::{RelayKnowledgeService, storage_api_error};
10
11impl RelayKnowledgeService {
12    pub async fn start_code_repository_watcher(&self) -> Result<Option<WatcherHandle>, ApiError> {
13        if !self.runtime.watcher.enabled {
14            self.stop_code_repository_watcher().await;
15            return Ok(None);
16        }
17
18        let mut guard = self.watcher.write().await;
19        if let Some(handle) = guard.as_ref() {
20            return Ok(Some(handle.clone()));
21        }
22
23        let store = self.store().await.map_err(storage_api_error)?;
24        let repositories = store
25            .list_code_repositories()
26            .await
27            .map_err(storage_api_error)?
28            .into_iter()
29            .filter_map(|status| watched_repository_from_status(&status))
30            .collect::<Vec<_>>();
31        let queue_store = Arc::clone(&store);
32        let handle = FileWatcher::new(self.runtime.watcher.clone())
33            .start_with_sink(repositories, move |seed| {
34                let store = Arc::clone(&queue_store);
35                async move {
36                    store
37                        .queue_code_index_task(seed)
38                        .await
39                        .map(|_| ())
40                        .map_err(|error| error.to_string())
41                }
42            })
43            .map_err(ApiError::storage_unavailable)?;
44        *guard = Some(handle.clone());
45
46        Ok(Some(handle))
47    }
48
49    pub async fn stop_code_repository_watcher(&self) {
50        let handle = self.watcher.write().await.take();
51        if let Some(handle) = handle {
52            handle.request_shutdown();
53        }
54    }
55
56    pub(crate) async fn refresh_watched_code_repository(
57        &self,
58        status: &CodeRepositoryStatus,
59    ) -> bool {
60        let Some(handle) = self.watcher.read().await.as_ref().cloned() else {
61            return false;
62        };
63        match watched_repository_from_status(status) {
64            Some(repository) => handle.add_repository(repository).await,
65            None => {
66                handle.remove_repository(&status.alias).await
67                    || handle.remove_repository(&status.repository_id).await
68            }
69        }
70    }
71
72    pub(crate) async fn remove_watched_code_repository(
73        &self,
74        alias: &str,
75        repository_id: &str,
76    ) -> bool {
77        let Some(handle) = self.watcher.read().await.as_ref().cloned() else {
78            return false;
79        };
80        handle.remove_repository(alias).await || handle.remove_repository(repository_id).await
81    }
82
83    pub(super) async fn watcher_diagnostics(&self) -> Option<crate::watcher::WatcherDiagnostics> {
84        self.watcher
85            .read()
86            .await
87            .as_ref()
88            .map(WatcherHandle::diagnostics)
89    }
90}
91
92fn watched_repository_from_status(status: &CodeRepositoryStatus) -> Option<WatchedRepository> {
93    if status.stale {
94        return None;
95    }
96    let source_scope = status.last_indexed_scope_id.clone()?;
97    Some(WatchedRepository {
98        repository_id: status.repository_id.clone(),
99        alias: status.alias.clone(),
100        root: PathBuf::from(status.root_path.clone()),
101        path_filters: status.path_filters.clone(),
102        language_filters: status.language_filters.clone(),
103        source_scope,
104    })
105}
106
107#[cfg(test)]
108mod tests {
109    use super::*;
110
111    fn status(last_indexed_scope_id: Option<&str>, stale: bool) -> CodeRepositoryStatus {
112        CodeRepositoryStatus {
113            repository_id: "repo-1".to_owned(),
114            alias: "core".to_owned(),
115            root_path: "/tmp/core".to_owned(),
116            path_filters: vec!["src".to_owned()],
117            language_filters: Vec::new(),
118            last_indexed_scope_id: last_indexed_scope_id.map(str::to_owned),
119            last_indexed_commit: None,
120            tree_hash: None,
121            state: "registered".to_owned(),
122            indexed_file_count: 0,
123            symbol_count: 0,
124            reference_count: 0,
125            chunk_count: 0,
126            stale,
127            degraded_reason: None,
128        }
129    }
130
131    #[test]
132    fn watched_repository_from_status_skips_unindexed_repositories() {
133        assert!(watched_repository_from_status(&status(None, true)).is_none());
134    }
135
136    #[test]
137    fn watched_repository_from_status_skips_stale_repositories() {
138        assert!(watched_repository_from_status(&status(Some("scope-1"), true)).is_none());
139    }
140
141    #[test]
142    fn watched_repository_from_status_uses_indexed_scope() {
143        let watched =
144            watched_repository_from_status(&status(Some("scope-1"), false)).expect("indexed repo");
145        assert_eq!(watched.repository_id, "repo-1");
146        assert_eq!(watched.alias, "core");
147        assert_eq!(watched.source_scope, "scope-1");
148        assert_eq!(watched.path_filters, vec!["src"]);
149    }
150}
relay_knowledge/application/service/watcher.rs

relay_knowledge/application/service/
watcher.rs
