pipedash_core/application/
refresh_manager.rs1use std::sync::Arc;
2use std::time::{
3 Duration,
4 Instant,
5};
6
7use serde::{
8 Deserialize,
9 Serialize,
10};
11use tokio::sync::{
12 Mutex,
13 RwLock,
14};
15use tokio::time::interval;
16
17use super::services::metrics_service::MetricsService;
18use super::services::pipeline_service::PipelineService;
19use crate::domain::Pipeline;
20use crate::event::{
21 CoreEvent,
22 EventBus,
23};
24
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
26#[serde(rename_all = "lowercase")]
27pub enum RefreshMode {
28 Active,
29 Idle,
30}
31
32impl RefreshMode {
33 pub fn as_str(&self) -> &'static str {
34 match self {
35 RefreshMode::Active => "active",
36 RefreshMode::Idle => "idle",
37 }
38 }
39}
40
41impl std::str::FromStr for RefreshMode {
42 type Err = std::convert::Infallible;
43
44 fn from_str(s: &str) -> Result<Self, Self::Err> {
45 Ok(match s.to_lowercase().as_str() {
46 "active" => RefreshMode::Active,
47 _ => RefreshMode::Idle,
48 })
49 }
50}
51
52pub struct RefreshManager {
53 pipeline_service: Arc<PipelineService>,
54 metrics_service: Option<Arc<MetricsService>>,
55 event_bus: Arc<dyn EventBus>,
56 mode: Arc<RwLock<RefreshMode>>,
57 running: Arc<RwLock<bool>>,
58 last_refresh: Arc<Mutex<Option<Instant>>>,
59 last_metrics_cleanup: Arc<Mutex<Option<Instant>>>,
60 no_change_count: Arc<Mutex<u32>>,
61 current_interval: Arc<Mutex<Duration>>,
62 priority_queue: Arc<Mutex<Vec<i64>>>,
63}
64
65impl RefreshManager {
66 pub fn new(
67 pipeline_service: Arc<PipelineService>, metrics_service: Option<Arc<MetricsService>>,
68 event_bus: Arc<dyn EventBus>,
69 ) -> Self {
70 Self {
71 pipeline_service,
72 metrics_service,
73 event_bus,
74 mode: Arc::new(RwLock::new(RefreshMode::Active)),
75 running: Arc::new(RwLock::new(false)),
76 last_refresh: Arc::new(Mutex::new(None)),
77 last_metrics_cleanup: Arc::new(Mutex::new(None)),
78 no_change_count: Arc::new(Mutex::new(0)),
79 current_interval: Arc::new(Mutex::new(Duration::from_secs(10))),
80 priority_queue: Arc::new(Mutex::new(Vec::new())),
81 }
82 }
83
84 pub async fn prioritize_provider(&self, provider_id: i64) {
85 let mut queue = self.priority_queue.lock().await;
86 if !queue.contains(&provider_id) {
87 queue.push(provider_id);
88 tracing::debug!(
89 provider_id = provider_id,
90 "Provider added to priority queue"
91 );
92 }
93 }
94
95 pub async fn start(&self) {
96 let mut running = self.running.write().await;
97 if *running {
98 return; }
100 *running = true;
101 drop(running);
102
103 let pipeline_service = Arc::clone(&self.pipeline_service);
104 let metrics_service = self.metrics_service.clone();
105 let event_bus = Arc::clone(&self.event_bus);
106 let mode = Arc::clone(&self.mode);
107 let running = Arc::clone(&self.running);
108 let last_refresh = Arc::clone(&self.last_refresh);
109 let last_metrics_cleanup = Arc::clone(&self.last_metrics_cleanup);
110 let no_change_count = Arc::clone(&self.no_change_count);
111 let current_interval = Arc::clone(&self.current_interval);
112 let priority_queue = Arc::clone(&self.priority_queue);
113
114 tokio::spawn(async move {
115 let mut tick_interval = interval(Duration::from_secs(5));
116
117 tracing::info!("RefreshManager started");
118
119 loop {
120 tick_interval.tick().await;
121
122 let is_running = *running.read().await;
123 if !is_running {
124 break;
125 }
126
127 let current_mode = *mode.read().await;
128
129 {
130 let mut queue = priority_queue.lock().await;
131 if !queue.is_empty() {
132 let providers_to_fetch: Vec<i64> = queue.drain(..).collect();
133
134 tracing::debug!(
135 count = providers_to_fetch.len(),
136 "Processing priority queue"
137 );
138
139 for provider_id in providers_to_fetch {
140 let service = Arc::clone(&pipeline_service);
141 tokio::spawn(async move {
142 if let Err(e) = service.fetch_pipelines(Some(provider_id)).await {
143 tracing::warn!(
144 provider_id = provider_id,
145 error = %e,
146 "Priority provider fetch failed"
147 );
148 }
149 });
150 }
151 }
152 }
153
154 if current_mode == RefreshMode::Active {
155 let refresh_interval = *current_interval.lock().await;
156 let should_refresh = {
157 let last = last_refresh.lock().await;
158 match *last {
159 Some(last_time) => last_time.elapsed() >= refresh_interval,
160 None => true,
161 }
162 };
163
164 if !should_refresh {
165 continue;
166 }
167
168 {
169 let mut last = last_refresh.lock().await;
170 *last = Some(Instant::now());
171 }
172
173 let old_cached = pipeline_service.get_cached_pipelines(None).await.ok();
174
175 match pipeline_service.fetch_pipelines(None).await {
176 Ok(pipelines) => {
177 let timestamp = std::time::SystemTime::now()
178 .duration_since(std::time::UNIX_EPOCH)
179 .unwrap()
180 .as_millis() as i64;
181
182 event_bus
183 .emit(CoreEvent::PipelinesUpdated {
184 pipelines: pipelines.clone(),
185 provider_id: None,
186 timestamp,
187 })
188 .await;
189
190 let has_changes = if let Some(cached) = old_cached {
191 let changes = Self::has_changes(&cached, &pipelines);
192 if changes {
193 event_bus
194 .emit(CoreEvent::PipelineStatusChanged {
195 pipelines: pipelines.clone(),
196 })
197 .await;
198
199 for new_pipeline in &pipelines {
200 if let Some(old_pipeline) =
201 cached.iter().find(|p| p.id == new_pipeline.id)
202 {
203 if old_pipeline.status != new_pipeline.status
204 || old_pipeline.last_run != new_pipeline.last_run
205 {
206 tracing::debug!(
207 pipeline = %new_pipeline.name,
208 "Pipeline changed, invalidating run cache"
209 );
210 pipeline_service
211 .invalidate_run_cache(&new_pipeline.id)
212 .await;
213 }
214 }
215 }
216 }
217 changes
218 } else {
219 false
220 };
221
222 let mut interval = current_interval.lock().await;
223 let mut count = no_change_count.lock().await;
224
225 if has_changes {
226 *interval = Duration::from_secs(10);
227 *count = 0;
228 if *interval > Duration::from_secs(10) {
229 tracing::debug!(
230 interval_secs = interval.as_secs(),
231 "Changes detected, decreased interval"
232 );
233 }
234 } else {
235 *count += 1;
236 if *count >= 3 && *interval < Duration::from_secs(300) {
237 *interval = Duration::from_secs((*interval).as_secs() * 2)
238 .min(Duration::from_secs(300));
239 tracing::debug!(
240 no_change_cycles = *count,
241 interval_secs = interval.as_secs(),
242 "No changes, increased interval"
243 );
244 }
245 }
246 }
247 Err(e) => {
248 let error_msg = format!("Refresh error: {e:?}");
249 event_bus
250 .emit(CoreEvent::RefreshError { error: error_msg })
251 .await;
252 }
253 }
254 }
255
256 if let Some(ref metrics_svc) = metrics_service {
257 let should_cleanup = {
258 let last = last_metrics_cleanup.lock().await;
259 match *last {
260 Some(last_time) => last_time.elapsed() >= Duration::from_secs(6 * 3600),
261 None => true,
262 }
263 };
264
265 if should_cleanup {
266 {
267 let mut last = last_metrics_cleanup.lock().await;
268 *last = Some(Instant::now());
269 }
270
271 match metrics_svc.cleanup_old_metrics().await {
272 Ok(deleted) => {
273 tracing::info!(deleted = deleted, "Metrics cleanup complete");
274 }
275 Err(e) => {
276 tracing::error!(error = %e, "Metrics cleanup failed");
277 }
278 }
279 }
280 }
281 }
282
283 tracing::info!("RefreshManager stopped");
284 });
285 }
286
287 pub async fn stop(&self) {
288 let mut running = self.running.write().await;
289 *running = false;
290 }
291
292 pub async fn get_mode(&self) -> RefreshMode {
293 *self.mode.read().await
294 }
295
296 pub async fn set_mode(&self, mode: RefreshMode) {
297 let mut current = self.mode.write().await;
298 *current = mode;
299 tracing::debug!(mode = mode.as_str(), "Refresh mode changed");
300 }
301
302 pub async fn is_running(&self) -> bool {
303 *self.running.read().await
304 }
305
306 pub async fn get_interval(&self) -> Duration {
307 *self.current_interval.lock().await
308 }
309
310 pub async fn reset_interval(&self) {
311 let mut interval = self.current_interval.lock().await;
312 let mut count = self.no_change_count.lock().await;
313 *interval = Duration::from_secs(10);
314 *count = 0;
315 }
316
317 fn has_changes(old: &[Pipeline], new: &[Pipeline]) -> bool {
318 if old.len() != new.len() {
319 return true;
320 }
321
322 for new_pipeline in new {
323 if let Some(old_pipeline) = old.iter().find(|p| p.id == new_pipeline.id) {
324 if old_pipeline.status != new_pipeline.status {
325 return true;
326 }
327 if old_pipeline.last_run != new_pipeline.last_run {
328 return true;
329 }
330 } else {
331 return true;
332 }
333 }
334
335 false
336 }
337}