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