clankerdiff_watch/
repository_watcher.rs

1use crate::{
2    error::WatchError,
3    file_watcher::{FileWatcher, NotifyFileWatcher, worktree_walk},
4    filter::should_refresh,
5};
6use clankerdiff_core::DiffScope;
7use clankerdiff_git::{GitError, GitRepository, RepositorySnapshot};
8use ignore::WalkBuilder;
9use std::{sync::Arc, time::Duration};
10use tokio::{
11    sync::{mpsc, oneshot, watch},
12    task::{JoinHandle, spawn_blocking},
13};
14
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub struct WatchOptions {
17    pub debounce: Duration,
18}
19
20impl Default for WatchOptions {
21    fn default() -> Self {
22        Self {
23            debounce: Duration::from_millis(150),
24        }
25    }
26}
27
28#[derive(Debug)]
29pub enum RepositoryRequest {
30    SetScope {
31        scope: DiffScope,
32        result_tx: oneshot::Sender<Result<(), Arc<GitError>>>,
33    },
34}
35
36/// The retained repository state: the last successful snapshot and the health of
37/// the most recent load. A failed load keeps the previous snapshot in place.
38#[derive(Debug, Clone)]
39pub struct RepositoryState {
40    pub snapshot: Arc<RepositorySnapshot>,
41    pub error: Option<Arc<GitError>>,
42}
43
44impl RepositoryState {
45    /// The most recent load failure as a display message, if any.
46    #[must_use]
47    pub fn error_message(&self) -> Option<String> {
48        self.error.as_ref().map(ToString::to_string)
49    }
50
51    /// Folds one load result in; returns whether anything observable changed.
52    fn apply(&mut self, result: Result<RepositorySnapshot, Arc<GitError>>) -> bool {
53        match result {
54            Ok(snapshot) => {
55                let recovered = self.error.take().is_some();
56                if *self.snapshot == snapshot {
57                    return recovered;
58                }
59                self.snapshot = Arc::new(snapshot);
60                true
61            }
62            Err(error) => {
63                if self.error_message() == Some(error.to_string()) {
64                    return false;
65                }
66                self.error = Some(error);
67                true
68            }
69        }
70    }
71}
72
73#[derive(Debug)]
74pub struct RepositoryWatcher {
75    pub request_tx: mpsc::Sender<RepositoryRequest>,
76    pub state_rx: watch::Receiver<RepositoryState>,
77    task: JoinHandle<()>,
78}
79
80impl RepositoryWatcher {
81    pub async fn spawn(
82        repository: GitRepository,
83        scope: DiffScope,
84        options: WatchOptions,
85    ) -> Result<Self, WatchError> {
86        let watcher = Self::create_file_watcher(&repository, options.debounce).await?;
87        let snapshot = repository.snapshot_with_sources(scope).await?;
88        let (request_tx, request_rx) = mpsc::channel(64);
89        let (state_tx, state_rx) = watch::channel(RepositoryState {
90            snapshot: Arc::new(snapshot),
91            error: None,
92        });
93        let actor = RepositoryActor {
94            repository,
95            watcher,
96            scope,
97            request_rx,
98            state_tx,
99        };
100        Ok(Self {
101            request_tx,
102            state_rx,
103            task: tokio::spawn(actor.run()),
104        })
105    }
106
107    async fn create_file_watcher(
108        repository: &GitRepository,
109        debounce: Duration,
110    ) -> Result<NotifyFileWatcher, WatchError> {
111        let directories = repository.metadata_directories().await?;
112        let mut worktree = worktree_walk(repository.root());
113        worktree.filter_entry(|entry| entry.file_name() != ".git");
114        let mut walks = vec![worktree];
115        if let Some((first, rest)) = directories.split_first() {
116            let mut metadata = WalkBuilder::new(first);
117            for directory in rest {
118                metadata.add(directory);
119            }
120            metadata.standard_filters(false).filter_entry(|entry| {
121                entry.depth() != 1 || matches!(entry.file_name().to_str(), Some("refs" | "info"))
122            });
123            walks.push(metadata);
124        }
125        let repository = repository.clone();
126        let watcher = spawn_blocking(move || {
127            NotifyFileWatcher::with_walks(walks, debounce, move |paths| {
128                let repository = repository.clone();
129                let directories = directories.clone();
130                async move { should_refresh(&repository, &directories, paths).await }
131            })
132        })
133        .await
134        .map_err(|_| WatchError::Stopped)??;
135        Ok(watcher)
136    }
137}
138
139impl Drop for RepositoryWatcher {
140    fn drop(&mut self) {
141        self.task.abort();
142    }
143}
144
145struct RepositoryActor {
146    repository: GitRepository,
147    watcher: NotifyFileWatcher,
148    scope: DiffScope,
149    request_rx: mpsc::Receiver<RepositoryRequest>,
150    state_tx: watch::Sender<RepositoryState>,
151}
152
153impl RepositoryActor {
154    async fn run(mut self) {
155        loop {
156            let result_tx = tokio::select! {
157                biased;
158                request = self.request_rx.recv() => match request {
159                    Some(RepositoryRequest::SetScope { scope, result_tx }) => {
160                        self.scope = scope;
161                        Some(result_tx)
162                    }
163                    None => return,
164                },
165                event = self.watcher.recv() => match event {
166                    Some(()) => None,
167                    None => return,
168                },
169            };
170
171            let result = self
172                .repository
173                .snapshot_with_sources(self.scope)
174                .await
175                .map_err(Arc::new);
176            let outcome = result.as_ref().map(|_| ()).map_err(Arc::clone);
177            self.state_tx.send_if_modified(|state| state.apply(result));
178
179            if let Some(tx) = result_tx {
180                let _ = tx.send(outcome);
181            }
182        }
183    }
184}
clankerdiff_watch/repository_watcher.rs

clankerdiff_watch/
repository_watcher.rs
