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 trigger_pipeline(&self, params: TriggerParams) -> DomainResult<String>;
215
216 async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> DomainResult<()>;
217
218 async fn review_run(
219 &self, pipeline_id: &str, run_number: i64, decision: pipedash_plugin_api::ReviewDecision,
220 ) -> DomainResult<()>;
221
222 async fn rerun_failed_jobs(&self, pipeline_id: &str, run_number: i64) -> DomainResult<()>;
223
224 async fn rerun_run(&self, pipeline_id: &str, run_number: i64) -> DomainResult<()>;
225
226 async fn get_workflow_parameters(
227 &self, workflow_id: &str,
228 ) -> DomainResult<Vec<pipedash_plugin_api::WorkflowParameter>>;
229
230 async fn fetch_rate_limit(&self) -> DomainResult<Option<pipedash_plugin_api::RateLimit>>;
231
232 #[allow(dead_code)]
233 async fn validate_credentials(&self) -> DomainResult<bool>;
234
235 #[allow(dead_code)]
236 fn provider_type(&self) -> &str;
237
238 fn capabilities(&self) -> &pipedash_plugin_api::PluginCapabilities;
239
240 fn renew_at(&self) -> Option<chrono::DateTime<chrono::Utc>> {
241 None
242 }
243
244 fn status_source(&self) -> Option<&pipedash_plugin_api::StatusSource>;
245
246 async fn check_health(&self) -> DomainResult<Vec<pipedash_plugin_api::ComponentHealth>>;
247
248 fn log_dialect(&self) -> &'static dyn LogDialect {
249 &PlainDialect
250 }
251
252 async fn fetch_log_nodes(&self, _run: &PipelineRun) -> DomainResult<Vec<LogNode>> {
253 Err(DomainError::NotSupported(
254 "Run logs are not supported by this provider".into(),
255 ))
256 }
257
258 async fn read_log(
259 &self, _run: &PipelineRun, _node_id: &str, _from: Option<&LogCursor>, _sink: &mut LogSink,
260 ) -> DomainResult<LogRead> {
261 Err(DomainError::NotSupported(
262 "Run logs are not supported by this provider".into(),
263 ))
264 }
265}
266
267#[cfg(test)]
268mod tests {
269 use chrono::{
270 DateTime,
271 Duration,
272 Utc,
273 };
274
275 use super::*;
276
277 fn now() -> DateTime<Utc> {
278 "2026-10-28T12:00:00Z".parse().unwrap()
279 }
280
281 fn compute(
282 status: FetchStatus, level: HealthLevel, fetched_ago: Option<Duration>,
283 paused_in: Option<Duration>, interval: i64,
284 ) -> ProviderFreshness {
285 ProviderFreshness::compute(
286 &status,
287 level,
288 fetched_ago.map(|ago| now() - ago),
289 paused_in.map(|ahead| now() + ahead),
290 interval,
291 now(),
292 )
293 }
294
295 fn plain(state: FreshnessState) -> ProviderFreshness {
296 ProviderFreshness {
297 state,
298 stale_at: None,
299 }
300 }
301
302 #[test]
303 fn error_status_is_down() {
304 let f = compute(
305 FetchStatus::Error,
306 HealthLevel::Operational,
307 Some(Duration::seconds(5)),
308 None,
309 30,
310 );
311 assert_eq!(f, plain(FreshnessState::Down));
312 }
313
314 #[test]
315 fn partial_and_major_outage_are_down_even_when_never_fetched() {
316 for level in [HealthLevel::PartialOutage, HealthLevel::MajorOutage] {
317 let f = compute(FetchStatus::Never, level, None, None, 30);
318 assert_eq!(f, plain(FreshnessState::Down));
319 }
320 let f = compute(
321 FetchStatus::Success,
322 HealthLevel::Degraded,
323 Some(Duration::seconds(5)),
324 None,
325 30,
326 );
327 assert_eq!(f.state, FreshnessState::Fresh);
328 }
329
330 #[test]
331 fn never_fetched_is_never() {
332 let f = compute(FetchStatus::Never, HealthLevel::Operational, None, None, 30);
333 assert_eq!(f, plain(FreshnessState::Never));
334 }
335
336 #[test]
337 fn paused_in_the_future_is_stale_but_past_pause_is_ignored() {
338 let paused = compute(
339 FetchStatus::Success,
340 HealthLevel::Operational,
341 Some(Duration::seconds(5)),
342 Some(Duration::seconds(30)),
343 30,
344 );
345 assert_eq!(paused, plain(FreshnessState::Stale));
346
347 let elapsed = compute(
348 FetchStatus::Success,
349 HealthLevel::Operational,
350 Some(Duration::seconds(5)),
351 Some(Duration::seconds(-30)),
352 30,
353 );
354 assert_eq!(elapsed.state, FreshnessState::Fresh);
355 }
356
357 #[test]
358 fn old_fetch_is_stale() {
359 let f = compute(
360 FetchStatus::Success,
361 HealthLevel::Operational,
362 Some(Duration::seconds(601)),
363 None,
364 30,
365 );
366 assert_eq!(f, plain(FreshnessState::Stale));
367 }
368
369 #[test]
370 fn threshold_is_the_larger_of_three_intervals_and_ten_minutes() {
371 let at_ten_minutes = compute(
372 FetchStatus::Success,
373 HealthLevel::Operational,
374 Some(Duration::seconds(600)),
375 None,
376 30,
377 );
378 assert_eq!(at_ten_minutes.state, FreshnessState::Fresh);
379
380 let long_interval_inside = compute(
381 FetchStatus::Success,
382 HealthLevel::Operational,
383 Some(Duration::seconds(1500)),
384 None,
385 600,
386 );
387 assert_eq!(long_interval_inside.state, FreshnessState::Fresh);
388
389 let long_interval_outside = compute(
390 FetchStatus::Success,
391 HealthLevel::Operational,
392 Some(Duration::seconds(1801)),
393 None,
394 600,
395 );
396 assert_eq!(long_interval_outside, plain(FreshnessState::Stale));
397 }
398
399 #[test]
400 fn fresh_reports_when_it_turns_stale() {
401 let f = compute(
402 FetchStatus::Success,
403 HealthLevel::Operational,
404 Some(Duration::seconds(100)),
405 None,
406 600,
407 );
408 assert_eq!(f.state, FreshnessState::Fresh);
409 assert_eq!(
410 f.stale_at,
411 Some(now() - Duration::seconds(100) + Duration::seconds(1800))
412 );
413
414 let short = compute(
415 FetchStatus::Success,
416 HealthLevel::Operational,
417 Some(Duration::seconds(100)),
418 None,
419 30,
420 );
421 assert_eq!(
422 short.stale_at,
423 Some(now() - Duration::seconds(100) + Duration::seconds(600))
424 );
425 }
426
427 #[test]
428 fn serializes_snake_case_and_defaults_to_never() {
429 let json = serde_json::to_value(ProviderFreshness::default()).unwrap();
430 assert_eq!(
431 json,
432 serde_json::json!({ "state": "never", "stale_at": null })
433 );
434 }
435
436 #[test]
437 fn recovery_asks_signed_in_providers_to_sign_in_again() {
438 let recovery = ProviderRecovery::for_failure;
439 assert_eq!(
440 recovery(Some(FailureKind::Auth), true),
441 Some(ProviderRecovery::SignIn)
442 );
443 assert_eq!(
444 recovery(Some(FailureKind::Auth), false),
445 Some(ProviderRecovery::Settings)
446 );
447 assert_eq!(
448 recovery(Some(FailureKind::Tls), true),
449 Some(ProviderRecovery::Settings)
450 );
451 assert_eq!(recovery(Some(FailureKind::Network), true), None);
452 assert_eq!(recovery(None, true), None);
453 }
454}