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