1use std::collections::HashMap;
2
3use async_trait::async_trait;
4use serde::{
5 Deserialize,
6 Serialize,
7};
8
9use super::credential::SignInCredential;
10use super::error::{
11 DomainError,
12 DomainResult,
13 FailureKind,
14};
15use super::health::{
16 HealthLevel,
17 ProviderHealth,
18};
19use super::pipeline::{
20 Pipeline,
21 PipelineRun,
22 TriggerParams,
23};
24use super::run_logs::{
25 LogCursor,
26 LogDialect,
27 LogNode,
28 LogRead,
29 LogSink,
30 PlainDialect,
31};
32
33#[derive(Debug, Clone, Serialize, Deserialize)]
34pub struct ProviderConfig {
35 pub id: Option<i64>,
36 pub name: String,
37 pub provider_type: String,
38 pub token: String,
39 pub config: HashMap<String, String>,
40 #[serde(default = "default_refresh_interval")]
41 pub refresh_interval: i64,
42 #[serde(default)]
43 pub version: Option<i64>,
44}
45
46fn default_refresh_interval() -> i64 {
47 30
48}
49
50impl ProviderConfig {
51 pub fn display_name(&self) -> &str {
52 self.config
53 .get("display_name")
54 .map(|s| s.as_str())
55 .unwrap_or(&self.name)
56 }
57
58 pub fn plugin_config(&self) -> HashMap<String, String> {
59 let mut config = self.config.clone();
60 config.insert("token".to_string(), SignInCredential::bearer(&self.token));
61 config
62 }
63}
64
65#[derive(Debug, Clone, Serialize, Deserialize)]
66#[serde(rename_all = "lowercase")]
67pub enum FetchStatus {
68 Success,
69 Error,
70 Never,
71}
72
73#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
74pub struct ProviderImportResult {
75 pub added: Vec<String>,
76 pub skipped: Vec<String>,
77}
78
79#[derive(Debug, Clone, Serialize, Deserialize)]
80pub struct ProviderSummary {
81 pub id: i64,
82 pub name: String,
83 pub provider_type: String,
84 pub icon: Option<String>,
85 pub pipeline_count: usize,
86 pub last_updated: Option<chrono::DateTime<chrono::Utc>>,
87 pub refresh_interval: i64,
88 pub configured_repositories: Vec<String>,
89 pub last_fetch_status: FetchStatus,
90 pub last_fetch_error: Option<String>,
91 pub last_fetch_at: Option<chrono::DateTime<chrono::Utc>>,
92 #[serde(default)]
93 pub last_success_at: Option<chrono::DateTime<chrono::Utc>>,
94 #[serde(default)]
95 pub paused_until: Option<chrono::DateTime<chrono::Utc>>,
96 #[serde(default)]
97 pub last_fetch_error_kind: Option<FailureKind>,
98 #[serde(default)]
99 pub health: ProviderHealth,
100 pub version: i64,
101 #[serde(default)]
102 pub freshness: ProviderFreshness,
103 #[serde(default)]
104 pub recovery: Option<ProviderRecovery>,
105}
106
107#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
108#[serde(rename_all = "snake_case")]
109pub enum ProviderRecovery {
110 SignIn,
111 Settings,
112}
113
114impl ProviderRecovery {
115 pub fn for_failure(kind: Option<FailureKind>, signed_in: bool) -> Option<Self> {
116 match kind? {
117 FailureKind::Auth if signed_in => Some(Self::SignIn),
118 FailureKind::Auth
119 | FailureKind::Permission
120 | FailureKind::Config
121 | FailureKind::Tls => Some(Self::Settings),
122 _ => None,
123 }
124 }
125}
126
127#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
128#[serde(rename_all = "snake_case")]
129pub enum FreshnessState {
130 Fresh,
131 Stale,
132 Down,
133 #[default]
134 Never,
135}
136
137#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
138pub struct ProviderFreshness {
139 pub state: FreshnessState,
140 pub stale_at: Option<chrono::DateTime<chrono::Utc>>,
141}
142
143const MIN_STALE_SECS: i64 = 10 * 60;
144
145impl ProviderFreshness {
146 pub fn compute(
147 status: &FetchStatus, level: HealthLevel,
148 last_fetch_at: Option<chrono::DateTime<chrono::Utc>>,
149 paused_until: Option<chrono::DateTime<chrono::Utc>>, refresh_interval: i64,
150 now: chrono::DateTime<chrono::Utc>,
151 ) -> Self {
152 let state = |state| Self {
153 state,
154 stale_at: None,
155 };
156 if matches!(status, FetchStatus::Error)
157 || matches!(level, HealthLevel::PartialOutage | HealthLevel::MajorOutage)
158 {
159 return state(FreshnessState::Down);
160 }
161 if matches!(status, FetchStatus::Never) {
162 return state(FreshnessState::Never);
163 }
164 let window =
165 chrono::TimeDelta::try_seconds(refresh_interval.saturating_mul(3).max(MIN_STALE_SECS))
166 .unwrap_or(chrono::TimeDelta::MAX);
167 let stale_at = last_fetch_at.and_then(|at| at.checked_add_signed(window));
168 if paused_until.is_some_and(|until| until > now) || stale_at.is_some_and(|at| now > at) {
169 return state(FreshnessState::Stale);
170 }
171 Self {
172 state: FreshnessState::Fresh,
173 stale_at,
174 }
175 }
176}
177
178#[async_trait]
179pub trait Provider: Send + Sync {
180 async fn fetch_pipelines(&self) -> DomainResult<Vec<Pipeline>>;
181
182 async fn fetch_pipelines_paginated(
183 &self, page: usize, page_size: usize,
184 ) -> DomainResult<pipedash_plugin_api::PaginatedResponse<Pipeline>> {
185 use pipedash_plugin_api::PaginatedResponse;
186
187 let all_pipelines = self.fetch_pipelines().await?;
188 let total_count = all_pipelines.len();
189 let start = (page - 1) * page_size;
190 let end = start + page_size;
191
192 let pipelines = if start < total_count {
193 all_pipelines[start..end.min(total_count)].to_vec()
194 } else {
195 Vec::new()
196 };
197
198 Ok(PaginatedResponse::new(
199 pipelines,
200 page,
201 page_size,
202 total_count,
203 ))
204 }
205
206 async fn fetch_run_history(
207 &self, pipeline_id: &str, limit: usize,
208 ) -> DomainResult<Vec<PipelineRun>>;
209
210 async fn fetch_run_details(
211 &self, pipeline_id: &str, run_number: i64,
212 ) -> DomainResult<PipelineRun>;
213
214 async fn fetch_recent_runs(
215 &self, _pipeline_ids: &[String], _limit: usize,
216 ) -> DomainResult<HashMap<String, Vec<PipelineRun>>> {
217 Err(DomainError::NotSupported(
218 "Batched run history is not supported by this provider".into(),
219 ))
220 }
221
222 async fn trigger_pipeline(&self, params: TriggerParams) -> DomainResult<String>;
223
224 async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> DomainResult<()>;
225
226 async fn review_run(
227 &self, pipeline_id: &str, run_number: i64, decision: pipedash_plugin_api::ReviewDecision,
228 ) -> DomainResult<()>;
229
230 async fn rerun_failed_jobs(&self, pipeline_id: &str, run_number: i64) -> DomainResult<()>;
231
232 async fn rerun_run(&self, pipeline_id: &str, run_number: i64) -> DomainResult<()>;
233
234 async fn enable_pipeline(&self, pipeline_id: &str) -> DomainResult<()>;
235
236 async fn get_workflow_parameters(
237 &self, workflow_id: &str,
238 ) -> DomainResult<Vec<pipedash_plugin_api::WorkflowParameter>>;
239
240 async fn fetch_rate_limit(&self) -> DomainResult<Option<pipedash_plugin_api::RateLimit>>;
241
242 #[allow(dead_code)]
243 async fn validate_credentials(&self) -> DomainResult<bool>;
244
245 #[allow(dead_code)]
246 fn provider_type(&self) -> &str;
247
248 fn capabilities(&self) -> &pipedash_plugin_api::PluginCapabilities;
249
250 fn renew_at(&self) -> Option<chrono::DateTime<chrono::Utc>> {
251 None
252 }
253
254 fn status_source(&self) -> Option<&pipedash_plugin_api::StatusSource>;
255
256 async fn check_health(&self) -> DomainResult<Vec<pipedash_plugin_api::ComponentHealth>>;
257
258 fn log_dialect(&self) -> &'static dyn LogDialect {
259 &PlainDialect
260 }
261
262 async fn fetch_log_nodes(&self, _run: &PipelineRun) -> DomainResult<Vec<LogNode>> {
263 Err(DomainError::NotSupported(
264 "Run logs are not supported by this provider".into(),
265 ))
266 }
267
268 async fn read_log(
269 &self, _run: &PipelineRun, _node_id: &str, _from: Option<&LogCursor>, _sink: &mut LogSink,
270 ) -> DomainResult<LogRead> {
271 Err(DomainError::NotSupported(
272 "Run logs are not supported by this provider".into(),
273 ))
274 }
275}
276
277#[cfg(test)]
278mod tests {
279 use chrono::{
280 DateTime,
281 Duration,
282 Utc,
283 };
284
285 use super::*;
286
287 fn now() -> DateTime<Utc> {
288 "2026-10-28T12:00:00Z".parse().unwrap()
289 }
290
291 fn compute(
292 status: FetchStatus, level: HealthLevel, fetched_ago: Option<Duration>,
293 paused_in: Option<Duration>, interval: i64,
294 ) -> ProviderFreshness {
295 ProviderFreshness::compute(
296 &status,
297 level,
298 fetched_ago.map(|ago| now() - ago),
299 paused_in.map(|ahead| now() + ahead),
300 interval,
301 now(),
302 )
303 }
304
305 fn plain(state: FreshnessState) -> ProviderFreshness {
306 ProviderFreshness {
307 state,
308 stale_at: None,
309 }
310 }
311
312 #[test]
313 fn error_status_is_down() {
314 let f = compute(
315 FetchStatus::Error,
316 HealthLevel::Operational,
317 Some(Duration::seconds(5)),
318 None,
319 30,
320 );
321 assert_eq!(f, plain(FreshnessState::Down));
322 }
323
324 #[test]
325 fn partial_and_major_outage_are_down_even_when_never_fetched() {
326 for level in [HealthLevel::PartialOutage, HealthLevel::MajorOutage] {
327 let f = compute(FetchStatus::Never, level, None, None, 30);
328 assert_eq!(f, plain(FreshnessState::Down));
329 }
330 let f = compute(
331 FetchStatus::Success,
332 HealthLevel::Degraded,
333 Some(Duration::seconds(5)),
334 None,
335 30,
336 );
337 assert_eq!(f.state, FreshnessState::Fresh);
338 }
339
340 #[test]
341 fn never_fetched_is_never() {
342 let f = compute(FetchStatus::Never, HealthLevel::Operational, None, None, 30);
343 assert_eq!(f, plain(FreshnessState::Never));
344 }
345
346 #[test]
347 fn paused_in_the_future_is_stale_but_past_pause_is_ignored() {
348 let paused = compute(
349 FetchStatus::Success,
350 HealthLevel::Operational,
351 Some(Duration::seconds(5)),
352 Some(Duration::seconds(30)),
353 30,
354 );
355 assert_eq!(paused, plain(FreshnessState::Stale));
356
357 let elapsed = compute(
358 FetchStatus::Success,
359 HealthLevel::Operational,
360 Some(Duration::seconds(5)),
361 Some(Duration::seconds(-30)),
362 30,
363 );
364 assert_eq!(elapsed.state, FreshnessState::Fresh);
365 }
366
367 #[test]
368 fn old_fetch_is_stale() {
369 let f = compute(
370 FetchStatus::Success,
371 HealthLevel::Operational,
372 Some(Duration::seconds(601)),
373 None,
374 30,
375 );
376 assert_eq!(f, plain(FreshnessState::Stale));
377 }
378
379 #[test]
380 fn threshold_is_the_larger_of_three_intervals_and_ten_minutes() {
381 let at_ten_minutes = compute(
382 FetchStatus::Success,
383 HealthLevel::Operational,
384 Some(Duration::seconds(600)),
385 None,
386 30,
387 );
388 assert_eq!(at_ten_minutes.state, FreshnessState::Fresh);
389
390 let long_interval_inside = compute(
391 FetchStatus::Success,
392 HealthLevel::Operational,
393 Some(Duration::seconds(1500)),
394 None,
395 600,
396 );
397 assert_eq!(long_interval_inside.state, FreshnessState::Fresh);
398
399 let long_interval_outside = compute(
400 FetchStatus::Success,
401 HealthLevel::Operational,
402 Some(Duration::seconds(1801)),
403 None,
404 600,
405 );
406 assert_eq!(long_interval_outside, plain(FreshnessState::Stale));
407 }
408
409 #[test]
410 fn fresh_reports_when_it_turns_stale() {
411 let f = compute(
412 FetchStatus::Success,
413 HealthLevel::Operational,
414 Some(Duration::seconds(100)),
415 None,
416 600,
417 );
418 assert_eq!(f.state, FreshnessState::Fresh);
419 assert_eq!(
420 f.stale_at,
421 Some(now() - Duration::seconds(100) + Duration::seconds(1800))
422 );
423
424 let short = compute(
425 FetchStatus::Success,
426 HealthLevel::Operational,
427 Some(Duration::seconds(100)),
428 None,
429 30,
430 );
431 assert_eq!(
432 short.stale_at,
433 Some(now() - Duration::seconds(100) + Duration::seconds(600))
434 );
435 }
436
437 #[test]
438 fn serializes_snake_case_and_defaults_to_never() {
439 let json = serde_json::to_value(ProviderFreshness::default()).unwrap();
440 assert_eq!(
441 json,
442 serde_json::json!({ "state": "never", "stale_at": null })
443 );
444 }
445
446 #[test]
447 fn recovery_asks_signed_in_providers_to_sign_in_again() {
448 let recovery = ProviderRecovery::for_failure;
449 assert_eq!(
450 recovery(Some(FailureKind::Auth), true),
451 Some(ProviderRecovery::SignIn)
452 );
453 assert_eq!(
454 recovery(Some(FailureKind::Auth), false),
455 Some(ProviderRecovery::Settings)
456 );
457 assert_eq!(
458 recovery(Some(FailureKind::Tls), true),
459 Some(ProviderRecovery::Settings)
460 );
461 assert_eq!(recovery(Some(FailureKind::Network), true), None);
462 assert_eq!(recovery(None, true), None);
463 }
464}