relay_knowledge/application/service/watcher/
mod.rs

1use std::{path::PathBuf, sync::Arc};
2
3use crate::{
4    api::ApiError,
5    domain::{CodeIndexTaskRecord, CodeIndexTaskState, 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                    let queued = store
37                        .queue_code_index_task(seed)
38                        .await
39                        .map_err(|error| error.to_string())?;
40                    accept_watcher_task(queued)
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 accept_watcher_task(task: CodeIndexTaskRecord) -> Result<(), String> {
93    if task.state != CodeIndexTaskState::DeadLetter {
94        return Ok(());
95    }
96    let failure = task
97        .last_error_message
98        .as_deref()
99        .map(|message| format!(": {message}"))
100        .unwrap_or_default();
101    Err(format!(
102        "durable code index task '{}' remains dead_letter{failure}; reset failed work before retrying repository '{}'",
103        task.task_id, task.alias
104    ))
105}
106
107fn watched_repository_from_status(status: &CodeRepositoryStatus) -> Option<WatchedRepository> {
108    if status.stale {
109        return None;
110    }
111    let source_scope = status.last_indexed_scope_id.clone()?;
112    let last_indexed_commit = status.last_indexed_commit.clone()?;
113    Some(WatchedRepository {
114        repository_id: status.repository_id.clone(),
115        alias: status.alias.clone(),
116        root: PathBuf::from(status.root_path.clone()),
117        path_filters: status.path_filters.clone(),
118        language_filters: status.language_filters.clone(),
119        source_scope,
120        last_indexed_commit,
121    })
122}
123
124#[cfg(test)]
125#[path = "mod_tests.rs"]
126mod tests;
relay_knowledge/application/service/watcher/mod.rs

relay_knowledge/application/service/watcher/
mod.rs
