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