1use std::collections::{
2 HashMap,
3 HashSet,
4};
5use std::future::Future;
6use std::sync::Arc;
7use std::time::Duration;
8
9use chrono::{
10 DateTime,
11 Utc,
12};
13use dashmap::DashMap;
14use pipedash_plugin_api::{
15 Plugin as PluginTrait,
16 PluginRegistry,
17 RateLimit,
18 CONTROLLER_HEALTH_FEATURE,
19};
20use tokio::sync::{
21 Mutex,
22 RwLock,
23};
24use tokio::time::{
25 timeout,
26 Instant,
27};
28
29use super::health_service::{
30 HealthHandle,
31 HealthInput,
32 ProviderHealthInput,
33};
34use super::sign_in_service::{
35 ProviderSignIn,
36 SignInService,
37};
38use crate::application::fetch_scheduler::{
39 FetchScheduler,
40 ForcedClaim,
41 Generation,
42};
43use crate::application::refresh_manager::RefreshDispatcher;
44use crate::domain::{
45 DomainError,
46 DomainResult,
47 FetchStatus,
48 HealthLevel,
49 Provider,
50 ProviderConfig,
51 ProviderFreshness,
52 ProviderHealth,
53 ProviderImportResult,
54 ProviderRecovery,
55 ProviderSummary,
56 SignInCredential,
57 SignInRef,
58 SignInStart,
59 SignInStatus,
60 SIGN_IN_CONFIG_KEY,
61};
62use crate::event::{
63 CoreEvent,
64 EventBus,
65};
66use crate::infrastructure::config::sync::{
67 ProviderSyncService,
68 SyncResult,
69};
70use crate::infrastructure::config::token_ref::TokenReference;
71use crate::infrastructure::config::{
72 ConfigLoader,
73 ConfigState,
74 GeneralSettingsUpdate,
75 PipedashConfig,
76 ProviderFileConfig,
77};
78use crate::infrastructure::database::{
79 ProviderFetchStatus,
80 Repository,
81 FETCH_STATUS_ERROR,
82 FETCH_STATUS_NEVER,
83 FETCH_STATUS_SUCCESS,
84};
85use crate::infrastructure::oauth;
86use crate::infrastructure::providers::PluginAdapter;
87use crate::infrastructure::run_logs::LogStore;
88use crate::infrastructure::{
89 HttpClientManager,
90 TokenStore,
91};
92use crate::plugins;
93mod discovery;
94
95const QUOTA_PAUSE_PERCENT: u64 = 10;
96const ACQUIRE_ATTEMPTS: usize = 3;
97
98pub(crate) type ProviderFactory =
99 Arc<dyn Fn(&ProviderConfig) -> DomainResult<Arc<dyn Provider>> + Send + Sync>;
100
101#[derive(Clone)]
102pub struct ProviderLease {
103 pub id: i64,
104 pub generation: Generation,
105 pub provider: Arc<dyn Provider>,
106}
107
108enum TakenSignIn {
109 None,
110 Flow(String),
111 Kept,
112}
113
114struct ProviderSlot {
115 generation: Generation,
116 provider: Arc<dyn Provider>,
117}
118
119impl ProviderSlot {
120 fn lease(&self, id: i64) -> ProviderLease {
121 ProviderLease {
122 id,
123 generation: self.generation,
124 provider: Arc::clone(&self.provider),
125 }
126 }
127}
128
129struct ParameterFetch<'a> {
130 fetches: &'a DashMap<String, Arc<Mutex<()>>>,
131 key: &'a str,
132 lock: Arc<Mutex<()>>,
133}
134
135impl<'a> ParameterFetch<'a> {
136 fn new(fetches: &'a DashMap<String, Arc<Mutex<()>>>, key: &'a str) -> Self {
137 let lock = Arc::clone(fetches.entry(key.to_string()).or_default().value());
138
139 Self { fetches, key, lock }
140 }
141}
142
143impl Drop for ParameterFetch<'_> {
144 fn drop(&mut self) {
145 self.fetches
146 .remove_if(self.key, |_, lock| Arc::ptr_eq(lock, &self.lock));
147 }
148}
149
150pub struct ProviderService {
151 repository: Arc<Repository>,
152 config_state: Arc<ConfigState>,
153 http_client_manager: Arc<HttpClientManager>,
154 providers: RwLock<HashMap<i64, ProviderSlot>>,
155 commit_locks: DashMap<i64, Arc<Mutex<()>>>,
156 renewal_locks: DashMap<i64, Arc<Mutex<()>>>,
157 sign_ins: Arc<SignInService>,
158 plugin_registry: Arc<PluginRegistry>,
159 parameter_fetches: DashMap<String, Arc<Mutex<()>>>,
160 discovery: discovery::DiscoveryCache,
161 event_bus: Arc<dyn EventBus>,
162 scheduler: Arc<FetchScheduler>,
163 health: HealthHandle,
164 dispatcher: RefreshDispatcher,
165 provider_factory: ProviderFactory,
166 config_apply: Mutex<()>,
167 log_store: Arc<LogStore>,
168}
169
170impl ProviderService {
171 #[allow(clippy::too_many_arguments)]
172 pub fn new(
173 repository: Arc<Repository>, http_client_manager: Arc<HttpClientManager>,
174 event_bus: Arc<dyn EventBus>, scheduler: Arc<FetchScheduler>, health: HealthHandle,
175 dispatcher: RefreshDispatcher, config_state: Arc<ConfigState>, log_store: Arc<LogStore>,
176 ) -> Self {
177 let plugin_registry = Arc::new(plugins::create_plugin_registry());
178 let provider_factory: ProviderFactory = {
179 let registry = Arc::clone(&plugin_registry);
180 let http = Arc::clone(&http_client_manager);
181 Arc::new(move |config| plugin_provider(®istry, &http, config))
182 };
183
184 Self {
185 repository,
186 config_state,
187 http_client_manager,
188 providers: RwLock::new(HashMap::new()),
189 commit_locks: DashMap::new(),
190 renewal_locks: DashMap::new(),
191 sign_ins: Arc::new(SignInService::default()),
192 plugin_registry,
193 parameter_fetches: DashMap::new(),
194 discovery: discovery::DiscoveryCache::default(),
195 event_bus,
196 scheduler,
197 health,
198 dispatcher,
199 provider_factory,
200 config_apply: Mutex::new(()),
201 log_store,
202 }
203 }
204
205 #[cfg(test)]
206 pub(crate) fn with_provider_factory(mut self, factory: ProviderFactory) -> Self {
207 self.provider_factory = factory;
208 self
209 }
210
211 pub fn repository(&self) -> &Arc<Repository> {
212 &self.repository
213 }
214
215 pub(crate) fn scheduler(&self) -> &Arc<FetchScheduler> {
216 &self.scheduler
217 }
218
219 pub(crate) fn health(&self) -> &HealthHandle {
220 &self.health
221 }
222
223 pub(crate) fn dispatcher(&self) -> &RefreshDispatcher {
224 &self.dispatcher
225 }
226
227 pub fn list_available_plugins(&self) -> Vec<pipedash_plugin_api::PluginMetadata> {
228 let mut metadata_list = Vec::new();
229 for provider_type in self.plugin_registry.provider_types() {
230 if let Some(plugin) = self.plugin_registry.get(&provider_type) {
231 metadata_list.push(plugin.metadata().clone());
232 }
233 }
234 metadata_list
235 }
236
237 pub fn retains_run_history(&self, provider_type: &str) -> bool {
238 self.plugin_registry
239 .get(provider_type)
240 .is_some_and(|plugin| plugin.metadata().capabilities.retains_run_history)
241 }
242
243 pub async fn init_plugin(
244 &self, provider_type: &str, mut config: HashMap<String, String>, provider_id: i64,
245 ) -> DomainResult<Box<dyn PluginTrait>> {
246 self.resolve_sign_in(&mut config).await?;
247 init_plugin(
248 &self.http_client_manager,
249 provider_type,
250 config,
251 provider_id,
252 )
253 }
254
255 pub async fn field_options(
256 &self, provider_type: &str, field_key: &str, mut config: HashMap<String, String>,
257 ) -> DomainResult<Vec<String>> {
258 self.resolve_sign_in(&mut config).await?;
259 let (mut plugin, client) = build_plugin(&self.http_client_manager, provider_type, &config)?;
260 let init_error = plugin.initialize(0, config.clone(), Some(client)).err();
261
262 match plugin.get_field_options(field_key, &config).await {
263 Ok(options) => Ok(options),
264 Err(e) => Err(match init_error {
265 Some(init) => {
266 DomainError::InvalidConfig(format!("Failed to initialize plugin: {init}"))
267 }
268 None => PluginAdapter::map_error(e),
269 }),
270 }
271 }
272
273 pub(crate) fn commit_lock(&self, id: i64) -> Arc<Mutex<()>> {
274 Arc::clone(self.commit_locks.entry(id).or_default().value())
275 }
276
277 fn renewal_lock(&self, id: i64) -> Arc<Mutex<()>> {
278 Arc::clone(self.renewal_locks.entry(id).or_default().value())
279 }
280
281 async fn resolve_sign_in(&self, config: &mut HashMap<String, String>) -> DomainResult<()> {
282 let Some(reference) = config.remove(SIGN_IN_CONFIG_KEY) else {
283 return Ok(());
284 };
285 let token = match SignInRef::parse(&reference) {
286 SignInRef::Flow(flow) => self.sign_ins.credential(&flow).await?.token,
287 SignInRef::Provider(id) => {
288 SignInCredential::bearer(&self.renewed_config(id).await?.token)
289 }
290 };
291 config.insert("token".to_string(), token);
292 Ok(())
293 }
294
295 async fn take_sign_in(
296 &self, config: &mut ProviderConfig, owner: Option<i64>,
297 ) -> DomainResult<TakenSignIn> {
298 let Some(reference) = config.config.remove(SIGN_IN_CONFIG_KEY) else {
299 return Ok(TakenSignIn::None);
300 };
301 match SignInRef::parse(&reference) {
302 SignInRef::Flow(flow) => {
303 self.sign_ins.check_owner(&flow, owner)?;
304 config.token = self.sign_ins.credential(&flow).await?.encode()?;
305 Ok(TakenSignIn::Flow(flow))
306 }
307 SignInRef::Provider(id) if Some(id) == owner => {
308 config.token = self.renewed_config(id).await?.token;
309 Ok(TakenSignIn::Kept)
310 }
311 SignInRef::Provider(_) => Err(DomainError::InvalidConfig(
312 "This sign-in belongs to another provider".into(),
313 )),
314 }
315 }
316
317 async fn renewed_config(&self, id: i64) -> DomainResult<ProviderConfig> {
318 let renewal = self.renewal_lock(id);
319 let _renewal = renewal.lock().await;
320 let config = self.repository.get_provider(id).await?;
321 let Some(current) = SignInCredential::decode(&config.token)
322 .filter(|credential| credential.renew_at().is_some_and(|at| at <= Utc::now()))
323 else {
324 return Ok(config);
325 };
326
327 let (plugin, http) = build_plugin(
328 &self.http_client_manager,
329 &config.provider_type,
330 &config.config,
331 )?;
332 let token = oauth::refresh(&http, ¤t, plugin.sign_in_refresh_rejections())
333 .await?
334 .encode()?;
335
336 let commit = self.commit_lock(id);
337 let _commit = commit.lock().await;
338 let mut latest = self.repository.get_provider(id).await?;
339 if latest.token != config.token {
340 return Ok(latest);
341 }
342 self.repository.store_token(id, &token).await?;
343 tracing::info!(provider_id = id, "Renewed the provider sign-in");
344 latest.token = token;
345 Ok(latest)
346 }
347
348 async fn renew(&self, stale: ProviderLease) -> DomainResult<ProviderLease> {
349 let config = self.renewed_config(stale.id).await?;
350 let provider = (self.provider_factory)(&config)?;
351 let mut providers = self.providers.write().await;
352 let slot = providers
353 .get_mut(&stale.id)
354 .ok_or_else(|| DomainError::ProviderNotFound(stale.id.to_string()))?;
355 if slot.generation == stale.generation {
356 slot.provider = provider;
357 }
358 Ok(slot.lease(stale.id))
359 }
360
361 pub async fn provider_for_editing(&self, id: i64) -> DomainResult<ProviderConfig> {
362 let mut config = self.repository.get_provider(id).await?;
363 if SignInCredential::decode(&config.token).is_some() {
364 config.token.clear();
365 config.config.insert(
366 SIGN_IN_CONFIG_KEY.to_string(),
367 SignInRef::Provider(id).to_string(),
368 );
369 }
370 Ok(config)
371 }
372
373 pub async fn stored_plugin(&self, id: i64) -> DomainResult<Box<dyn PluginTrait>> {
374 let config = self.renewed_config(id).await?;
375 init_plugin(
376 &self.http_client_manager,
377 &config.provider_type,
378 config.plugin_config(),
379 id,
380 )
381 }
382
383 pub async fn start_sign_in(
384 &self, provider_type: &str, config: HashMap<String, String>,
385 ) -> DomainResult<SignInStart> {
386 self.start_sign_in_with_owner(provider_type, config, None)
387 .await
388 }
389
390 pub async fn start_provider_sign_in(&self, id: i64) -> DomainResult<SignInStart> {
391 let config = self.repository.get_provider(id).await?;
392 self.start_sign_in_with_owner(
393 &config.provider_type,
394 config.config,
395 Some((id, config.token)),
396 )
397 .await
398 }
399
400 async fn start_sign_in_with_owner(
401 &self, provider_type: &str, config: HashMap<String, String>, owner: Option<(i64, String)>,
402 ) -> DomainResult<SignInStart> {
403 let (plugin, http) = build_plugin(&self.http_client_manager, provider_type, &config)?;
404 let client = plugin
405 .sign_in_client(&config, &http)
406 .await
407 .map_err(PluginAdapter::map_error)?;
408 let owner = owner.map(|(id, token)| ProviderSignIn {
409 id,
410 provider_type: provider_type.to_string(),
411 token,
412 client: client.clone(),
413 });
414 self.sign_ins
415 .start(client, http, plugin.sign_in_refresh_rejections(), owner)
416 .await
417 }
418
419 pub async fn reconnect_provider(&self, id: i64, flow: &str) -> DomainResult<()> {
420 let owner = self.sign_ins.provider_flow(flow, id)?;
421 let token = self.sign_ins.credential(flow).await?.encode()?;
422
423 for _ in 0..ACQUIRE_ATTEMPTS {
424 let config = self.repository.get_provider(id).await?;
425 if config.provider_type != owner.provider_type || config.token != owner.token {
426 return Err(DomainError::ConcurrentModification(
427 "This provider's authentication changed. Start signing in again.".into(),
428 ));
429 }
430 let (plugin, http) = build_plugin(
431 &self.http_client_manager,
432 &config.provider_type,
433 &config.config,
434 )?;
435 let client = plugin
436 .sign_in_client(&config.config, &http)
437 .await
438 .map_err(PluginAdapter::map_error)?;
439 if client != owner.client {
440 return Err(DomainError::ConcurrentModification(
441 "This provider's sign-in settings changed. Start signing in again.".into(),
442 ));
443 }
444
445 let lock = self.commit_lock(id);
446 let guard = lock.lock().await;
447 self.sign_ins.check_owner(flow, Some(id))?;
448 let mut latest = self.repository.get_provider(id).await?;
449 if latest.provider_type != config.provider_type
450 || latest.config != config.config
451 || latest.token != config.token
452 {
453 continue;
454 }
455 latest.token.clone_from(&token);
456 let provider = (self.provider_factory)(&latest)?;
457 self.repository.store_token(id, &token).await?;
458 if let Err(e) = self.repository.reset_provider_fetch_status(id).await {
459 tracing::warn!(provider_id = id, error = %e, "Failed to reset fetch status");
460 }
461 let lease = self.replace(id, provider, latest.refresh_interval).await;
462 self.health.send(HealthInput::ProvidersChanged {
463 changed: vec![self.health_input(&latest, &lease)],
464 });
465 match self.summary_of(id).await {
466 Ok(provider) => {
467 self.event_bus
468 .emit(CoreEvent::ProviderUpdated {
469 provider,
470 timestamp: Utc::now().timestamp_millis(),
471 })
472 .await;
473 }
474 Err(e) => {
475 tracing::warn!(provider_id = id, error = %e, "Failed to build updated provider summary");
476 self.event_bus.emit(CoreEvent::ProvidersChanged).await;
477 }
478 }
479 self.sign_ins.forget(flow);
480 drop(guard);
481 self.queue_fetch(&lease);
482 return Ok(());
483 }
484
485 Err(DomainError::ConcurrentModification(format!(
486 "Provider {id} kept changing while it was reconnected"
487 )))
488 }
489
490 pub async fn sign_in_status(&self, reference: &str) -> DomainResult<SignInStatus> {
491 match SignInRef::parse(reference) {
492 SignInRef::Flow(flow) => Ok(self.sign_ins.status(&flow)),
493 SignInRef::Provider(id) => match self.renewed_config(id).await {
494 Ok(config) => Ok(SignInCredential::decode(&config.token).map_or(
495 SignInStatus::Expired,
496 |credential| SignInStatus::Completed {
497 account: credential.account,
498 },
499 )),
500 Err(DomainError::ProviderAuthenticationFailed(message)) => {
501 Ok(SignInStatus::Failed { message })
502 }
503 Err(e) => Err(e),
504 },
505 }
506 }
507
508 pub async fn complete_sign_in(
509 &self, flow: &str, redirect_url: &str,
510 ) -> DomainResult<SignInStatus> {
511 self.sign_ins.complete(flow, redirect_url).await
512 }
513
514 pub async fn cancel_sign_in(&self, flow: &str) {
515 self.sign_ins.cancel(flow).await;
516 }
517
518 pub async fn wait_for_sign_in(&self, flow: &str) -> SignInStatus {
519 self.sign_ins.wait(flow).await
520 }
521
522 async fn purge_run_logs(&self, id: i64) {
523 if let Err(e) = self.log_store.purge_provider(id).await {
524 tracing::warn!(provider_id = id, error = %e, "Failed to purge cached run logs");
525 }
526 }
527
528 async fn current(&self, id: i64) -> Option<ProviderLease> {
529 self.providers
530 .read()
531 .await
532 .get(&id)
533 .map(|slot| slot.lease(id))
534 }
535
536 pub(crate) async fn generation_of(&self, id: i64) -> Option<Generation> {
537 self.providers
538 .read()
539 .await
540 .get(&id)
541 .map(|slot| slot.generation)
542 }
543
544 pub(crate) async fn health_inputs(&self) -> Vec<ProviderHealthInput> {
545 let leases: Vec<ProviderLease> = self
546 .providers
547 .read()
548 .await
549 .iter()
550 .map(|(id, slot)| slot.lease(*id))
551 .collect();
552 let mut inputs = Vec::with_capacity(leases.len());
553 for lease in leases {
554 match self.repository.get_provider(lease.id).await {
555 Ok(config) => inputs.push(self.health_input(&config, &lease)),
556 Err(e) => {
557 tracing::debug!(provider_id = lease.id, error = %e, "Provider has no config row");
558 }
559 }
560 }
561
562 inputs
563 }
564
565 async fn install(&self, id: i64, provider: Arc<dyn Provider>, interval: i64) -> ProviderLease {
566 let mut providers = self.providers.write().await;
567 providers
568 .entry(id)
569 .or_insert_with(|| {
570 let generation = Generation::next();
571 self.scheduler
572 .register(id, generation, interval_of(interval));
573 ProviderSlot {
574 generation,
575 provider,
576 }
577 })
578 .lease(id)
579 }
580
581 async fn replace(&self, id: i64, provider: Arc<dyn Provider>, interval: i64) -> ProviderLease {
582 let generation = Generation::next();
583 let mut providers = self.providers.write().await;
584 providers.insert(
585 id,
586 ProviderSlot {
587 generation,
588 provider: Arc::clone(&provider),
589 },
590 );
591 self.scheduler.reset(id, generation, interval_of(interval));
592
593 ProviderLease {
594 id,
595 generation,
596 provider,
597 }
598 }
599
600 pub async fn acquire(&self, id: i64) -> DomainResult<ProviderLease> {
601 if let Some(lease) = self.current(id).await {
602 if lease.provider.renew_at().is_some_and(|at| at <= Utc::now()) {
603 return self.renew(lease).await;
604 }
605 return Ok(lease);
606 }
607
608 for _ in 0..ACQUIRE_ATTEMPTS {
609 let config = self.renewed_config(id).await?;
610 let provider = (self.provider_factory)(&config)?;
611 let lock = self.commit_lock(id);
612 let _guard = lock.lock().await;
613 if let Some(lease) = self.current(id).await {
614 return Ok(lease);
615 }
616 let row = self.repository.get_provider(id).await?;
617 if row.version == config.version {
618 let lease = self.install(id, provider, row.refresh_interval).await;
619 self.health.send(HealthInput::ProvidersChanged {
620 changed: vec![self.health_input(&row, &lease)],
621 });
622 return Ok(lease);
623 }
624 }
625
626 Err(DomainError::ConcurrentModification(format!(
627 "Provider {id} kept changing while it was loaded"
628 )))
629 }
630
631 pub async fn commit<F, Fut, T>(&self, lease: &ProviderLease, f: F) -> Option<T>
632 where
633 F: FnOnce() -> Fut,
634 Fut: Future<Output = T>,
635 {
636 self.commit_at(lease.id, lease.generation, f).await
637 }
638
639 async fn commit_at<F, Fut, T>(&self, id: i64, generation: Generation, f: F) -> Option<T>
640 where
641 F: FnOnce() -> Fut,
642 Fut: Future<Output = T>,
643 {
644 let lock = self.commit_lock(id);
645 let _guard = lock.lock().await;
646 let current = self
647 .providers
648 .read()
649 .await
650 .get(&id)
651 .map(|slot| slot.generation);
652 if current != Some(generation) {
653 return None;
654 }
655
656 Some(f().await)
657 }
658
659 pub async fn publish_summary(&self, id: i64, generation: Generation) {
660 self.commit_at(id, generation, || self.emit_status(id))
661 .await;
662 }
663
664 pub(crate) async fn emit_status(&self, id: i64) {
665 match self.summary_of(id).await {
666 Ok(provider) => {
667 self.event_bus
668 .emit(CoreEvent::ProviderStatusUpdated { provider })
669 .await;
670 }
671 Err(e) => {
672 tracing::warn!(provider_id = id, error = %e, "Failed to build provider summary");
673 }
674 }
675 }
676
677 async fn summary_of(&self, id: i64) -> DomainResult<ProviderSummary> {
678 let config = self.repository.get_provider(id).await?;
679
680 self.build_summary(id, &config).await
681 }
682
683 async fn build_summary(
684 &self, id: i64, config: &ProviderConfig,
685 ) -> DomainResult<ProviderSummary> {
686 let cached_pipelines = self.repository.get_cached_pipelines(Some(id)).await?;
687 let fetch = match self.repository.get_provider_fetch_status(id).await {
688 Ok(fetch) => fetch,
689 Err(e) => {
690 tracing::warn!(provider_id = id, error = %e, "Failed to read provider fetch status");
691 ProviderFetchStatus {
692 status: FETCH_STATUS_NEVER.to_string(),
693 error: None,
694 fetched_at: None,
695 succeeded_at: None,
696 }
697 }
698 };
699 let last_fetch_status = match fetch.status.as_str() {
700 FETCH_STATUS_SUCCESS => FetchStatus::Success,
701 FETCH_STATUS_ERROR => FetchStatus::Error,
702 _ => FetchStatus::Never,
703 };
704 let now = Instant::now();
705 let wall = Utc::now();
706 let view = self.scheduler.view(id, now);
707 let generation = self
708 .providers
709 .read()
710 .await
711 .get(&id)
712 .map(|slot| slot.generation);
713 let snapshot = self.health.snapshot();
714 let state = snapshot
715 .providers
716 .get(&id)
717 .filter(|state| Some(state.generation) == generation);
718 let paused_until = self.scheduler.paused_until(id, wall);
719 let level = state.map_or(HealthLevel::Operational, |state| state.level);
720 let freshness = ProviderFreshness::compute(
721 &last_fetch_status,
722 level,
723 fetch.fetched_at,
724 paused_until,
725 config.refresh_interval,
726 wall,
727 );
728
729 let last_fetch_error_kind = view.as_ref().and_then(|view| view.last_failure);
730 let recovery = ProviderRecovery::for_failure(
731 matches!(last_fetch_status, FetchStatus::Error)
732 .then_some(last_fetch_error_kind)
733 .flatten(),
734 SignInCredential::decode(&config.token).is_some(),
735 );
736
737 Ok(ProviderSummary {
738 id,
739 name: config.display_name().to_string(),
740 provider_type: config.provider_type.clone(),
741 icon: self
742 .plugin_registry
743 .get(&config.provider_type)
744 .and_then(|plugin| plugin.metadata().icon.clone()),
745 pipeline_count: cached_pipelines.len(),
746 last_updated: cached_pipelines.iter().map(|p| p.last_updated).max(),
747 refresh_interval: config.refresh_interval,
748 configured_repositories: configured_repositories(config),
749 last_fetch_status,
750 last_fetch_error: fetch.error,
751 last_fetch_at: fetch.fetched_at,
752 last_success_at: fetch.succeeded_at,
753 paused_until,
754 last_fetch_error_kind,
755 health: ProviderHealth {
756 level,
757 signals: state.map(|state| state.signals.clone()).unwrap_or_default(),
758 next_attempt_at: view
759 .as_ref()
760 .filter(|view| view.failures > 0)
761 .and_then(|_| {
762 let floor = state
763 .filter(|state| state.outage_floor)
764 .map(|_| self.scheduler.policy().outage_floor);
765 self.scheduler.next_due_in(id, now, floor)
766 })
767 .and_then(|wait| chrono::Duration::from_std(wait).ok())
768 .map(|wait| wall + wait),
769 in_flight: view.as_ref().is_some_and(|view| view.in_flight),
770 status_page: state.and_then(|state| state.status_page.clone()),
771 },
772 version: config.version.unwrap_or(1),
773 freshness,
774 recovery,
775 })
776 }
777
778 fn health_input(&self, config: &ProviderConfig, lease: &ProviderLease) -> ProviderHealthInput {
779 ProviderHealthInput {
780 id: lease.id,
781 generation: Some(lease.generation),
782 display_name: config.display_name().to_string(),
783 source: lease.provider.status_source().cloned(),
784 has_instance_checks: controller_health(&self.plugin_registry, &config.provider_type),
785 }
786 }
787
788 fn queue_fetch(&self, lease: &ProviderLease) {
789 if let ForcedClaim::Claimed(claim) =
790 self.scheduler
791 .claim_forced(lease.id, lease.generation, Instant::now())
792 {
793 self.dispatcher.fetch_now(vec![claim]);
794 }
795 }
796
797 pub async fn add_provider(&self, mut config: ProviderConfig) -> DomainResult<i64> {
798 let sign_in = self.take_sign_in(&mut config, None).await?;
799 let plugin = self
800 .init_plugin(&config.provider_type, config.plugin_config(), 0)
801 .await?;
802
803 let (validation_result, permission_status) =
804 tokio::join!(plugin.validate_credentials(), plugin.check_permissions());
805
806 validation_result.map_err(|e| {
807 DomainError::InvalidConfig(format!("Credential validation failed: {e}"))
808 })?;
809
810 let provider_type = plugin.provider_type();
811 if provider_type != config.provider_type {
812 return Err(DomainError::InvalidConfig(format!(
813 "Provider type mismatch: expected '{}', got '{}'",
814 config.provider_type, provider_type
815 )));
816 }
817
818 keep_selection_with_ci(plugin.as_ref(), &mut config).await?;
819
820 let permission_status = permission_status.ok();
821
822 let id = self.repository.add_provider(&config).await?;
823
824 if let Some(status) = &permission_status {
825 if let Err(e) = self.repository.store_provider_permissions(id, status).await {
826 tracing::warn!(provider_id = id, error = %e, "Failed to store permissions");
827 }
828 }
829
830 let mut config_with_id = config.clone();
831 config_with_id.id = Some(id);
832 let provider = (self.provider_factory)(&config_with_id)?;
833
834 let lock = self.commit_lock(id);
835 let _guard = lock.lock().await;
836 let lease = self.install(id, provider, config.refresh_interval).await;
837 self.health.send(HealthInput::ProvidersChanged {
838 changed: vec![self.health_input(&config_with_id, &lease)],
839 });
840 let provider = self.summary_of(id).await?;
841 self.event_bus
842 .emit(CoreEvent::ProviderAdded {
843 provider,
844 timestamp: Utc::now().timestamp_millis(),
845 })
846 .await;
847 if let TakenSignIn::Flow(flow) = sign_in {
848 self.sign_ins.forget(&flow);
849 }
850
851 Ok(id)
852 }
853
854 pub async fn export_providers_toml(&self) -> DomainResult<String> {
855 let providers = self.repository.config_backend().list_providers().await?;
856
857 ProviderSyncService::export_toml(&providers)
858 }
859
860 pub async fn import_providers_toml(&self, content: &str) -> DomainResult<ProviderImportResult> {
861 let file = ConfigLoader::parse_raw(content).map_err(|e| {
862 DomainError::InvalidConfig(format!("Couldn't read the providers file: {e}"))
863 })?;
864 if file.providers.is_empty() {
865 return Err(DomainError::InvalidConfig(
866 "The file has no [providers] entries".to_string(),
867 ));
868 }
869 if let Some((key, provider)) = file
870 .providers
871 .iter()
872 .find(|(_, provider)| !self.plugin_registry.is_registered(&provider.provider_type))
873 {
874 return Err(DomainError::InvalidProviderType(format!(
875 "Provider '{key}' has an unknown type '{}'",
876 provider.provider_type
877 )));
878 }
879
880 let existing: HashSet<String> = self
881 .repository
882 .config_backend()
883 .list_providers()
884 .await?
885 .into_iter()
886 .map(|provider| provider.name)
887 .collect();
888 let mut result = ProviderImportResult::default();
889 let mut failure = None;
890
891 for (key, provider) in &file.providers {
892 let name = provider.display_name(key).to_string();
893 if existing.contains(key) {
894 result.skipped.push(name);
895 continue;
896 }
897
898 let config = ProviderSyncService::provider_config(key, provider, String::new());
899 match self.repository.add_provider(&config).await {
900 Ok(_) => result.added.push(name),
901 Err(e) => {
902 let added_elsewhere = matches!(e, DomainError::InvalidConfig(_))
903 && self
904 .repository
905 .config_backend()
906 .list_providers()
907 .await
908 .is_ok_and(|providers| providers.iter().any(|p| p.name == *key));
909 if !added_elsewhere {
910 failure = Some(e);
911 break;
912 }
913 result.skipped.push(name);
914 }
915 }
916 }
917
918 if !result.added.is_empty() {
919 self.event_bus.emit(CoreEvent::ProvidersChanged).await;
920 }
921
922 failure.map_or(Ok(result), Err)
923 }
924
925 pub async fn apply_config_file(&self, config_state: &ConfigState) -> DomainResult<SyncResult> {
926 let _apply = self.config_apply.lock().await;
927 let PipedashConfig {
928 providers: file,
929 views,
930 ..
931 } = config_state.load_file()?;
932 let current = config_state.get().await;
933 let changed: Vec<(String, ProviderFileConfig)> = file
934 .iter()
935 .filter(|(key, provider)| current.providers.get(*key) != Some(*provider))
936 .map(|(key, provider)| (key.clone(), provider.clone()))
937 .collect();
938
939 for (key, provider) in &changed {
940 if !self.plugin_registry.is_registered(&provider.provider_type) {
941 return Err(DomainError::InvalidProviderType(format!(
942 "Provider '{key}' has an unknown type '{}'",
943 provider.provider_type
944 )));
945 }
946 (self.provider_factory)(&ProviderSyncService::provider_config(
947 key,
948 provider,
949 provider.token.clone(),
950 ))?;
951 }
952
953 let mut result = SyncResult::default();
954 if !changed.is_empty() {
955 let existing = self.repository.config_backend().list_providers().await?;
956 let token_store = config_state.token_store();
957 for (key, provider) in &changed {
958 let id = existing.iter().find(|p| p.name == *key).and_then(|p| p.id);
959 self.apply_config_provider(key, provider, id, token_store.as_ref())
960 .await?;
961 config_state.remember_provider(key, provider).await;
962 match id {
963 Some(_) => result.updated.push(key.clone()),
964 None => result.added.push(key.clone()),
965 }
966 }
967
968 tracing::info!(
969 added = result.added.len(),
970 updated = result.updated.len(),
971 "Applied provider changes from config file"
972 );
973 self.event_bus.emit(CoreEvent::ProvidersChanged).await;
974 }
975 if current.views != views {
976 config_state.remember_views(views).await;
977 self.event_bus.emit(CoreEvent::PipelineViewsChanged).await;
978 }
979 config_state.remember_providers(file).await;
980
981 Ok(result)
982 }
983
984 pub async fn update_general_settings(
985 &self, config_state: &ConfigState, update: GeneralSettingsUpdate,
986 ) -> DomainResult<()> {
987 let proxy_url = update.proxy_url.clone();
988 config_state.update_general_settings(update).await?;
989 if self.http_client_manager.set_proxy(&proxy_url)? {
990 self.rebuild_loaded_providers().await;
991 }
992
993 Ok(())
994 }
995
996 async fn rebuild_loaded_providers(&self) {
997 let ids: Vec<i64> = self.providers.read().await.keys().copied().collect();
998 let mut changed = Vec::new();
999
1000 for id in ids {
1001 let lock = self.commit_lock(id);
1002 let _guard = lock.lock().await;
1003 if !self.providers.read().await.contains_key(&id) {
1004 continue;
1005 }
1006
1007 let rebuilt = match self.repository.get_provider(id).await {
1008 Ok(config) => (self.provider_factory)(&config).map(|provider| (config, provider)),
1009 Err(e) => Err(e),
1010 };
1011 match rebuilt {
1012 Ok((config, provider)) => {
1013 let lease = self.replace(id, provider, config.refresh_interval).await;
1014 changed.push(self.health_input(&config, &lease));
1015 }
1016 Err(e) => {
1017 tracing::warn!(provider_id = id, error = %e, "Failed to rebuild provider");
1018 }
1019 }
1020 }
1021
1022 if !changed.is_empty() {
1023 self.health.send(HealthInput::ProvidersChanged { changed });
1024 }
1025 }
1026
1027 async fn apply_config_provider(
1028 &self, key: &str, provider: &ProviderFileConfig, existing: Option<i64>,
1029 token_store: &dyn TokenStore,
1030 ) -> DomainResult<()> {
1031 let sync = ProviderSyncService::new();
1032 let backend = self.repository.config_backend().as_ref();
1033 let id = match existing {
1034 Some(id) => id,
1035 None => sync.create_from_toml(key, provider, backend).await?,
1036 };
1037
1038 let lock = self.commit_lock(id);
1039 let guard = lock.lock().await;
1040 if existing.is_some() {
1041 let row = backend
1042 .get_provider(id)
1043 .await?
1044 .ok_or_else(|| DomainError::ProviderNotFound(id.to_string()))?;
1045 sync.update_from_toml(key, provider, &row, backend).await?;
1046 }
1047 ProviderSyncService::store_resolved_token(
1048 id,
1049 &provider.token,
1050 existing.is_none(),
1051 token_store,
1052 )
1053 .await?;
1054
1055 let config = self.repository.get_provider(id).await?;
1056 let instance = (self.provider_factory)(&config)?;
1057 if let Err(e) = self.repository.reset_provider_fetch_status(id).await {
1058 tracing::warn!(provider_id = id, error = %e, "Failed to reset fetch status");
1059 }
1060 let lease = self.replace(id, instance, config.refresh_interval).await;
1061 self.purge_run_logs(id).await;
1062 self.health.send(HealthInput::ProvidersChanged {
1063 changed: vec![self.health_input(&config, &lease)],
1064 });
1065 drop(guard);
1066
1067 self.queue_fetch(&lease);
1068
1069 Ok(())
1070 }
1071
1072 pub async fn get_provider_config(&self, id: i64) -> DomainResult<ProviderConfig> {
1073 self.repository.get_provider(id).await
1074 }
1075
1076 pub async fn list_providers(&self) -> DomainResult<Vec<ProviderSummary>> {
1077 let configs = self.repository.list_providers().await?;
1078 let mut summaries = Vec::with_capacity(configs.len());
1079
1080 for config in configs {
1081 if let Some(id) = config.id {
1082 summaries.push(self.build_summary(id, &config).await?);
1083 }
1084 }
1085
1086 Ok(summaries)
1087 }
1088
1089 pub async fn update_provider(&self, id: i64, mut config: ProviderConfig) -> DomainResult<()> {
1090 let saved_name = self.repository.get_provider(id).await?.name;
1091 if self
1092 .config_state
1093 .get()
1094 .await
1095 .providers
1096 .contains_key(&saved_name)
1097 {
1098 let display_name = std::mem::replace(&mut config.name, saved_name);
1099 config
1100 .config
1101 .insert("display_name".to_string(), display_name);
1102 }
1103 let sign_in = self.take_sign_in(&mut config, Some(id)).await?;
1104 let plugin = self
1105 .init_plugin(&config.provider_type, config.plugin_config(), id)
1106 .await?;
1107
1108 plugin.validate_credentials().await.map_err(|e| {
1109 DomainError::InvalidConfig(format!("Credential validation failed: {e}"))
1110 })?;
1111
1112 keep_selection_with_ci(plugin.as_ref(), &mut config).await?;
1113
1114 let permission_status = plugin.check_permissions().await.ok();
1115
1116 let lock = self.commit_lock(id);
1117 let guard = lock.lock().await;
1118
1119 let current = self.repository.get_provider(id).await?;
1120 if matches!(sign_in, TakenSignIn::Kept) {
1121 config.token.clone_from(¤t.token);
1122 }
1123 let mut config_with_id = config.clone();
1124 config_with_id.id = Some(id);
1125 let new_provider = (self.provider_factory)(&config_with_id)?;
1126 let success = self
1127 .repository
1128 .update_provider_with_version(id, &config, current.version.unwrap_or(1))
1129 .await?;
1130
1131 if !success {
1132 return Err(DomainError::ConcurrentModification(format!(
1133 "Provider {} was modified by another client",
1134 id
1135 )));
1136 }
1137
1138 if let Err(e) = self.repository.reset_provider_fetch_status(id).await {
1139 tracing::warn!(provider_id = id, error = %e, "Failed to reset fetch status");
1140 }
1141 let lease = self
1142 .replace(id, new_provider, config.refresh_interval)
1143 .await;
1144 self.purge_run_logs(id).await;
1145 self.health.send(HealthInput::ProvidersChanged {
1146 changed: vec![self.health_input(&config_with_id, &lease)],
1147 });
1148 match self.summary_of(id).await {
1149 Ok(provider) => {
1150 self.event_bus
1151 .emit(CoreEvent::ProviderUpdated {
1152 provider,
1153 timestamp: Utc::now().timestamp_millis(),
1154 })
1155 .await;
1156 }
1157 Err(e) => {
1158 tracing::warn!(provider_id = id, error = %e, "Failed to build updated provider summary");
1159 self.event_bus.emit(CoreEvent::ProvidersChanged).await;
1160 }
1161 }
1162 drop(guard);
1163
1164 if let Some(status) = &permission_status {
1165 if let Err(e) = self.repository.store_provider_permissions(id, status).await {
1166 tracing::warn!(provider_id = id, error = %e, "Failed to store permissions");
1167 }
1168 }
1169
1170 self.queue_fetch(&lease);
1171 if let TakenSignIn::Flow(flow) = sign_in {
1172 self.sign_ins.forget(&flow);
1173 }
1174
1175 Ok(())
1176 }
1177
1178 pub async fn update_provider_refresh_interval(
1179 &self, id: i64, refresh_interval: i64,
1180 ) -> DomainResult<()> {
1181 let lock = self.commit_lock(id);
1182 let _guard = lock.lock().await;
1183
1184 let mut config = self.repository.get_provider(id).await?;
1185 config.refresh_interval = refresh_interval;
1186 self.repository.update_provider(id, &config).await?;
1187
1188 let generation = self
1189 .providers
1190 .read()
1191 .await
1192 .get(&id)
1193 .map(|slot| slot.generation);
1194 if let Some(generation) = generation {
1195 self.scheduler
1196 .set_interval(id, generation, interval_of(refresh_interval));
1197 }
1198
1199 Ok(())
1200 }
1201
1202 pub async fn remove_provider(&self, id: i64) -> DomainResult<()> {
1203 let _apply = self.config_apply.lock().await;
1204 let lock = self.commit_lock(id);
1205 let _guard = lock.lock().await;
1206
1207 let config = self.repository.get_provider(id).await?;
1208 let provider = self.build_summary(id, &config).await?;
1209
1210 self.repository
1211 .remove_provider(id, &self.config_state)
1212 .await?;
1213 self.providers.write().await.remove(&id);
1214 self.purge_run_logs(id).await;
1215 self.scheduler.remove(id);
1216 self.health.send(HealthInput::ProvidersChanged {
1217 changed: vec![ProviderHealthInput {
1218 id,
1219 generation: None,
1220 display_name: config.display_name().to_string(),
1221 source: None,
1222 has_instance_checks: false,
1223 }],
1224 });
1225 self.event_bus
1226 .emit(CoreEvent::ProviderRemoved {
1227 provider,
1228 timestamp: Utc::now().timestamp_millis(),
1229 })
1230 .await;
1231
1232 Ok(())
1233 }
1234
1235 pub async fn get_provider(&self, id: i64) -> DomainResult<Arc<dyn Provider>> {
1236 Ok(self.acquire(id).await?.provider)
1237 }
1238
1239 pub async fn check_quota(&self, lease: &ProviderLease) -> Option<DateTime<Utc>> {
1240 let rate_limit =
1241 match timeout(Duration::from_secs(10), lease.provider.fetch_rate_limit()).await {
1242 Ok(Ok(rate_limit)) => rate_limit,
1243 Ok(Err(e)) => {
1244 tracing::debug!(provider_id = lease.id, error = %e, "Rate limit check failed");
1245 None
1246 }
1247 Err(_) => {
1248 tracing::debug!(provider_id = lease.id, "Rate limit check timed out");
1249 None
1250 }
1251 };
1252 let paused_until = rate_limit.and_then(|rate_limit| quota_pause(&rate_limit, Utc::now()));
1253
1254 self.commit(lease, || async {
1255 let changed = self.scheduler.set_quota_pause(
1256 lease.id,
1257 lease.generation,
1258 paused_until,
1259 Instant::now(),
1260 Utc::now(),
1261 );
1262 if let (true, Some(until)) = (changed, paused_until) {
1263 tracing::info!(provider_id = lease.id, %until, "API quota almost used up, pausing provider");
1264 }
1265 if changed {
1266 self.emit_status(lease.id).await;
1267 }
1268
1269 paused_until
1270 })
1271 .await
1272 .flatten()
1273 }
1274
1275 pub async fn get_provider_instance(&self, id: i64) -> DomainResult<Arc<dyn Provider>> {
1276 self.get_provider(id).await
1277 }
1278
1279 pub async fn load_missing_providers(&self) {
1280 let ids = match self.repository.list_provider_ids().await {
1281 Ok(ids) => ids,
1282 Err(e) => {
1283 tracing::warn!(error = %e, "Failed to list providers");
1284 return;
1285 }
1286 };
1287 let loaded: Vec<i64> = self.providers.read().await.keys().copied().collect();
1288 for id in ids.into_iter().filter(|id| !loaded.contains(id)) {
1289 if let Err(e) = self.acquire(id).await {
1290 tracing::debug!(provider_id = id, error = %e, "Provider not loaded yet");
1291 }
1292 }
1293 }
1294
1295 pub async fn load_all_providers(&self) -> DomainResult<()> {
1296 let configs = self.repository.list_providers().await?;
1297 let mut changed = Vec::new();
1298
1299 for config in configs {
1300 let Some(id) = config.id else {
1301 continue;
1302 };
1303 if config.token.is_empty() {
1304 tracing::debug!(
1305 provider_id = id,
1306 provider_type = %config.provider_type,
1307 "Skipping provider load - token not available yet (vault initializing in background)"
1308 );
1309 continue;
1310 }
1311
1312 match (self.provider_factory)(&config) {
1313 Ok(provider) => {
1314 let lock = self.commit_lock(id);
1315 let _guard = lock.lock().await;
1316 let lease = self.replace(id, provider, config.refresh_interval).await;
1317 changed.push(self.health_input(&config, &lease));
1318 }
1319 Err(e) => {
1320 tracing::warn!(provider_id = id, error = %e, "Failed to load provider");
1321 }
1322 }
1323 }
1324
1325 if !changed.is_empty() {
1326 self.health.send(HealthInput::ProvidersChanged { changed });
1327 }
1328
1329 Ok(())
1330 }
1331
1332 pub async fn get_workflow_parameters(
1333 &self, provider_id: i64, workflow_id: &str,
1334 ) -> DomainResult<Vec<pipedash_plugin_api::WorkflowParameter>> {
1335 if let Some(cached) = self.cached_parameters(workflow_id).await {
1336 return Ok(cached);
1337 }
1338
1339 let lease = self.acquire(provider_id).await?;
1340 let fetch = ParameterFetch::new(&self.parameter_fetches, workflow_id);
1341 let _guard = fetch.lock.lock().await;
1342
1343 if let Some(cached) = self.cached_parameters(workflow_id).await {
1344 return Ok(cached);
1345 }
1346
1347 let parameters = lease.provider.get_workflow_parameters(workflow_id).await?;
1348
1349 let cached = self
1350 .commit(&lease, || {
1351 self.repository
1352 .cache_workflow_parameters(workflow_id, ¶meters)
1353 })
1354 .await;
1355 if let Some(Err(e)) = cached {
1356 tracing::warn!(provider_id, workflow_id, error = %e, "Failed to cache workflow parameters");
1357 }
1358
1359 Ok(parameters)
1360 }
1361
1362 async fn cached_parameters(
1363 &self, workflow_id: &str,
1364 ) -> Option<Vec<pipedash_plugin_api::WorkflowParameter>> {
1365 match self
1366 .repository
1367 .get_cached_workflow_parameters(workflow_id)
1368 .await
1369 {
1370 Ok(cached) => cached,
1371 Err(e) => {
1372 tracing::warn!(workflow_id, error = %e, "Failed to read cached workflow parameters");
1373 None
1374 }
1375 }
1376 }
1377
1378 pub async fn get_provider_permissions(
1379 &self, provider_id: i64,
1380 ) -> DomainResult<Option<pipedash_plugin_api::PermissionStatus>> {
1381 self.repository.get_provider_permissions(provider_id).await
1382 }
1383
1384 pub async fn recheck_provider_permissions(
1385 &self, provider_id: i64,
1386 ) -> DomainResult<pipedash_plugin_api::PermissionStatus> {
1387 let plugin = self.stored_plugin(provider_id).await?;
1388
1389 let status = plugin
1390 .check_permissions()
1391 .await
1392 .map_err(|e| DomainError::InvalidConfig(format!("Failed to check permissions: {e}")))?;
1393
1394 self.repository
1395 .store_provider_permissions(provider_id, &status)
1396 .await?;
1397
1398 Ok(status)
1399 }
1400}
1401
1402fn interval_of(seconds: i64) -> Duration {
1403 Duration::from_secs(u64::try_from(seconds).unwrap_or(0))
1404}
1405
1406fn build_plugin(
1407 http: &HttpClientManager, provider_type: &str, config: &HashMap<String, String>,
1408) -> DomainResult<(Box<dyn PluginTrait>, Arc<reqwest::Client>)> {
1409 let plugin = plugins::create_plugin(provider_type).ok_or_else(|| {
1410 DomainError::InvalidProviderType(format!("Unknown provider type: {}", provider_type))
1411 })?;
1412 let client = http.client_for(&plugin.client_options(config))?;
1413
1414 Ok((plugin, client))
1415}
1416
1417async fn keep_selection_with_ci(
1418 plugin: &dyn PluginTrait, config: &mut ProviderConfig,
1419) -> DomainResult<()> {
1420 let Some(selected) = config.config.get("selected_items") else {
1421 return Ok(());
1422 };
1423 let ids: Vec<String> = selected
1424 .split(',')
1425 .map(str::trim)
1426 .filter(|id| !id.is_empty())
1427 .map(str::to_owned)
1428 .collect();
1429 if ids.is_empty() {
1430 return Ok(());
1431 }
1432 let Ok(kept) = plugin.ids_with_ci(ids.clone()).await else {
1433 return Ok(());
1434 };
1435 if kept.is_empty() {
1436 return Err(DomainError::InvalidConfig(
1437 "None of the selected repositories has a CI file or a pipeline run yet".to_string(),
1438 ));
1439 }
1440 if kept.len() < ids.len() {
1441 tracing::info!(
1442 selected = ids.len(),
1443 kept = kept.len(),
1444 "Skipped selected repositories without a CI file or a pipeline run"
1445 );
1446 config
1447 .config
1448 .insert("selected_items".to_string(), kept.join(","));
1449 }
1450 Ok(())
1451}
1452
1453fn init_plugin(
1454 http: &HttpClientManager, provider_type: &str, config: HashMap<String, String>,
1455 provider_id: i64,
1456) -> DomainResult<Box<dyn PluginTrait>> {
1457 let (mut plugin, client) = build_plugin(http, provider_type, &config)?;
1458 plugin
1459 .initialize(provider_id, config, Some(client))
1460 .map_err(|e| DomainError::InvalidConfig(format!("Failed to initialize plugin: {e}")))?;
1461
1462 Ok(plugin)
1463}
1464
1465pub(crate) fn controller_health(registry: &PluginRegistry, provider_type: &str) -> bool {
1466 registry.get(provider_type).is_some_and(|plugin| {
1467 plugin
1468 .metadata()
1469 .features
1470 .iter()
1471 .any(|feature| feature.id == CONTROLLER_HEALTH_FEATURE)
1472 })
1473}
1474
1475pub(crate) fn plugin_provider(
1476 registry: &PluginRegistry, http: &HttpClientManager, config: &ProviderConfig,
1477) -> DomainResult<Arc<dyn Provider>> {
1478 if !registry.is_registered(&config.provider_type) {
1479 return Err(DomainError::InvalidProviderType(format!(
1480 "Plugin not found for provider type: {}",
1481 config.provider_type
1482 )));
1483 }
1484
1485 let provider_id = config.id.unwrap_or(0);
1486 let (token, renew_at) = match SignInCredential::decode(&config.token) {
1487 Some(credential) => {
1488 let renew_at = credential.renew_at();
1489 (credential.token, renew_at)
1490 }
1491 None => (resolve_token_reference(provider_id, &config.token), None),
1492 };
1493 let mut plugin_config = config.config.clone();
1494 plugin_config.insert("token".to_string(), token);
1495
1496 let plugin = init_plugin(http, &config.provider_type, plugin_config, provider_id)?;
1497
1498 Ok(Arc::new(PluginAdapter::new(
1499 plugin,
1500 config.provider_type.clone(),
1501 provider_id,
1502 renew_at,
1503 )))
1504}
1505
1506fn resolve_token_reference(provider_id: i64, token: &str) -> String {
1507 if token.is_empty() {
1508 return String::new();
1509 }
1510 match TokenReference::parse(token) {
1511 Ok(TokenReference::EnvVar(var_name)) => std::env::var(&var_name).unwrap_or_else(|_| {
1512 tracing::warn!(
1513 provider_id = provider_id,
1514 var_name = %var_name,
1515 "Environment variable not found, using empty token"
1516 );
1517 String::new()
1518 }),
1519 Ok(TokenReference::None) => String::new(),
1520 Ok(_) | Err(_) => token.to_string(),
1521 }
1522}
1523
1524fn configured_repositories(config: &ProviderConfig) -> Vec<String> {
1525 config
1526 .config
1527 .get("selected_items")
1528 .map(|items| {
1529 if matches!(
1530 config.provider_type.as_str(),
1531 "argocd" | "fluxcd" | "komodo" | "prow"
1532 ) {
1533 return Vec::new();
1534 }
1535
1536 items
1537 .split(',')
1538 .map(|s| s.trim().to_string())
1539 .filter(|s| !s.is_empty())
1540 .map(|item| {
1541 if config.provider_type == "tekton" && item.contains("__") {
1542 item.replace("__", "/")
1543 } else if config.provider_type == "jenkins" && !item.contains('/') {
1544 format!("(root)/{}", item)
1545 } else {
1546 item
1547 }
1548 })
1549 .collect()
1550 })
1551 .unwrap_or_default()
1552}
1553
1554pub(crate) fn quota_pause(rate_limit: &RateLimit, now: DateTime<Utc>) -> Option<DateTime<Utc>> {
1555 let low = rate_limit.remaining * 100 < rate_limit.limit * QUOTA_PAUSE_PERCENT;
1556
1557 (low && rate_limit.reset_at > now).then_some(rate_limit.reset_at)
1558}
1559
1560#[cfg(test)]
1561mod tests {
1562 use std::io::{
1563 Read,
1564 Write,
1565 };
1566
1567 use pipedash_plugin_api::PluginError;
1568
1569 use super::*;
1570 use crate::application::test_support::{
1571 pipeline,
1572 FeedStub,
1573 Gate,
1574 Harness,
1575 StubProvider,
1576 };
1577 use crate::application::FetchResult;
1578 use crate::domain::PipelineStatus;
1579 use crate::infrastructure::config::ConfigLoader;
1580 use crate::infrastructure::oauth::test_idp::{
1581 jwt,
1582 Idp,
1583 };
1584 use crate::test_env::EnvGuard;
1585
1586 const GITHUB_SECRET: &str = "ghp_exampleSECRET0123456789";
1587 const GITLAB_SECRET: &str = "glpat-SECRET-VALUE";
1588 const JENKINS_DEFAULT_SECRET: &str = "hunter2-default";
1589
1590 fn provider(
1591 name: &str, provider_type: &str, token: &str, refresh_interval: i64,
1592 config: &[(&str, &str)],
1593 ) -> ProviderConfig {
1594 ProviderConfig {
1595 id: None,
1596 name: name.to_string(),
1597 provider_type: provider_type.to_string(),
1598 token: token.to_string(),
1599 config: config
1600 .iter()
1601 .map(|(k, v)| (k.to_string(), v.to_string()))
1602 .collect(),
1603 refresh_interval,
1604 version: None,
1605 }
1606 }
1607
1608 struct CiCheck(Option<&'static [&'static str]>);
1609
1610 #[async_trait::async_trait]
1611 impl PluginTrait for CiCheck {
1612 fn metadata(&self) -> &pipedash_plugin_api::PluginMetadata {
1613 unimplemented!()
1614 }
1615
1616 fn initialize(
1617 &mut self, _: i64, _: HashMap<String, String>, _: Option<Arc<reqwest::Client>>,
1618 ) -> pipedash_plugin_api::PluginResult<()> {
1619 Ok(())
1620 }
1621
1622 async fn validate_credentials(&self) -> pipedash_plugin_api::PluginResult<bool> {
1623 Ok(true)
1624 }
1625
1626 async fn fetch_pipelines(
1627 &self,
1628 ) -> pipedash_plugin_api::PluginResult<Vec<pipedash_plugin_api::Pipeline>> {
1629 Ok(Vec::new())
1630 }
1631
1632 async fn fetch_run_history(
1633 &self, _: &str, _: usize,
1634 ) -> pipedash_plugin_api::PluginResult<Vec<pipedash_plugin_api::PipelineRun>> {
1635 Ok(Vec::new())
1636 }
1637
1638 async fn fetch_run_details(
1639 &self, _: &str, _: i64,
1640 ) -> pipedash_plugin_api::PluginResult<pipedash_plugin_api::PipelineRun> {
1641 Err(PluginError::NotSupported(String::new()))
1642 }
1643
1644 async fn trigger_pipeline(
1645 &self, _: pipedash_plugin_api::TriggerParams,
1646 ) -> pipedash_plugin_api::PluginResult<String> {
1647 Err(PluginError::NotSupported(String::new()))
1648 }
1649
1650 async fn ids_with_ci(
1651 &self, _: Vec<String>,
1652 ) -> pipedash_plugin_api::PluginResult<Vec<String>> {
1653 self.0
1654 .map(|kept| kept.iter().map(|id| id.to_string()).collect())
1655 .ok_or_else(|| PluginError::NetworkError("reset".to_string()))
1656 }
1657 }
1658
1659 #[tokio::test]
1660 async fn a_saved_selection_keeps_only_repositories_with_ci() {
1661 let selected = &[("selected_items", "acme/api, acme/empty,acme/web")];
1662
1663 let mut config = provider("gh", "github", "t", 30, selected);
1664 keep_selection_with_ci(&CiCheck(Some(&["acme/api", "acme/web"])), &mut config)
1665 .await
1666 .unwrap();
1667 assert_eq!(config.config["selected_items"], "acme/api,acme/web");
1668
1669 let mut config = provider("gh", "github", "t", 30, selected);
1670 keep_selection_with_ci(&CiCheck(None), &mut config)
1671 .await
1672 .unwrap();
1673 assert_eq!(
1674 config.config["selected_items"],
1675 "acme/api, acme/empty,acme/web"
1676 );
1677
1678 let mut config = provider("gh", "github", "t", 30, selected);
1679 let error = keep_selection_with_ci(&CiCheck(Some(&[])), &mut config)
1680 .await
1681 .unwrap_err();
1682 assert!(matches!(error, DomainError::InvalidConfig(_)), "{error}");
1683 assert_eq!(
1684 config.config["selected_items"],
1685 "acme/api, acme/empty,acme/web"
1686 );
1687 }
1688
1689 #[tokio::test]
1690 async fn test_export_providers_toml_round_trips_without_tokens() {
1691 let source = Harness::with_real_plugins().await;
1692 let repository = source.repository.clone();
1693 let service = &source.provider_service;
1694
1695 let originals = [
1696 provider(
1697 "github-main",
1698 "github",
1699 GITHUB_SECRET,
1700 45,
1701 &[("selected_items", "acme/api,acme/web")],
1702 ),
1703 provider(
1704 "My Jenkins",
1705 "jenkins",
1706 &format!("${{JENKINS_API_TOKEN:-{JENKINS_DEFAULT_SECRET}}}"),
1707 120,
1708 &[
1709 ("display_name", "Prod Jenkins"),
1710 ("server_url", "http://127.0.0.1:9"),
1711 ("username", "ci"),
1712 ("selected_items", "(root)/deploy"),
1713 ],
1714 ),
1715 provider("gitlab_main", "gitlab", GITLAB_SECRET, 30, &[]),
1716 ];
1717 for original in &originals {
1718 repository.add_provider(original).await.unwrap();
1719 }
1720
1721 let exported = service.export_providers_toml().await.unwrap();
1722
1723 for secret in [GITHUB_SECRET, GITLAB_SECRET, JENKINS_DEFAULT_SECRET] {
1724 assert!(!exported.contains(secret), "export leaked {secret}");
1725 }
1726 assert!(exported.contains("token = \"${PIPEDASH_TOKEN_GITHUB_MAIN:-}\""));
1727 assert!(exported.contains("token = \"${JENKINS_API_TOKEN:-}\""));
1728 assert!(exported.contains("token = \"${PIPEDASH_TOKEN_GITLAB_MAIN:-}\""));
1729
1730 let _env = EnvGuard::set(&[
1731 ("PIPEDASH_TOKEN_GITHUB_MAIN", Some("env-github-token")),
1732 ("JENKINS_API_TOKEN", Some("env-jenkins-token")),
1733 ("PIPEDASH_TOKEN_GITLAB_MAIN", None),
1734 ]);
1735 let parsed = ConfigLoader::parse(&exported).unwrap();
1736
1737 let mut keys: Vec<_> = parsed.providers.keys().map(String::as_str).collect();
1738 keys.sort_unstable();
1739 assert_eq!(keys, ["My Jenkins", "github-main", "gitlab_main"]);
1740 assert_eq!(parsed.providers["github-main"].token, "env-github-token");
1741 assert_eq!(parsed.providers["My Jenkins"].token, "env-jenkins-token");
1742 assert_eq!(parsed.providers["gitlab_main"].token, "");
1743 assert_eq!(
1744 parsed.providers["My Jenkins"].name.as_deref(),
1745 Some("Prod Jenkins")
1746 );
1747
1748 let target_harness = Harness::with_real_plugins().await;
1749 let target = &target_harness.repository;
1750 crate::infrastructure::config::sync::ProviderSyncService::new()
1751 .sync_toml_to_db(
1752 parsed.providers,
1753 target.config_backend().as_ref(),
1754 &crate::infrastructure::MemoryTokenStore::new(),
1755 false,
1756 )
1757 .await
1758 .unwrap();
1759
1760 let imported = target.config_backend().list_providers().await.unwrap();
1761
1762 assert_eq!(imported.len(), originals.len());
1763 for original in &originals {
1764 let imported = imported.iter().find(|p| p.name == original.name).unwrap();
1765 assert_eq!(imported.name, original.name);
1766 assert_eq!(imported.provider_type, original.provider_type);
1767 assert_eq!(imported.refresh_interval, original.refresh_interval);
1768 assert_eq!(imported.config, original.config);
1769 }
1770 }
1771
1772 #[tokio::test]
1773 async fn test_import_providers_toml_adds_missing_providers_without_tokens() {
1774 let source = Harness::with_real_plugins().await;
1775 let jenkins = provider(
1776 "My Jenkins",
1777 "jenkins",
1778 "${JENKINS_API_TOKEN}",
1779 120,
1780 &[
1781 ("display_name", "Prod Jenkins"),
1782 ("server_url", "http://127.0.0.1:9"),
1783 ("username", "ci"),
1784 ("selected_items", "(root)/deploy"),
1785 ],
1786 );
1787 for original in [
1788 provider(
1789 "github-main",
1790 "github",
1791 GITHUB_SECRET,
1792 45,
1793 &[("selected_items", "acme/api")],
1794 ),
1795 jenkins.clone(),
1796 ] {
1797 source.repository.add_provider(&original).await.unwrap();
1798 }
1799 let exported = source
1800 .provider_service
1801 .export_providers_toml()
1802 .await
1803 .unwrap();
1804
1805 let target = Harness::with_real_plugins().await;
1806 let kept = provider(
1807 "github-main",
1808 "github",
1809 "target-token",
1810 300,
1811 &[("selected_items", "acme/web")],
1812 );
1813 let kept_id = target.repository.add_provider(&kept).await.unwrap();
1814
1815 let _env = EnvGuard::set(&[("JENKINS_API_TOKEN", Some("env-jenkins-token"))]);
1816 let result = target
1817 .provider_service
1818 .import_providers_toml(&exported)
1819 .await
1820 .unwrap();
1821
1822 assert_eq!(
1823 result,
1824 ProviderImportResult {
1825 added: vec!["Prod Jenkins".to_string()],
1826 skipped: vec!["github-main".to_string()],
1827 }
1828 );
1829 let unchanged = target.repository.get_provider(kept_id).await.unwrap();
1830 assert_eq!(unchanged.token, kept.token);
1831 assert_eq!(unchanged.refresh_interval, kept.refresh_interval);
1832 assert_eq!(unchanged.config, kept.config);
1833
1834 let imported = target.repository.list_providers().await.unwrap();
1835 let imported = imported.iter().find(|p| p.name == jenkins.name).unwrap();
1836 let row = target
1837 .repository
1838 .config_backend()
1839 .get_provider(imported.id.unwrap())
1840 .await
1841 .unwrap()
1842 .unwrap();
1843 assert_eq!(imported.token, "");
1844 assert_eq!(row.token, "");
1845 assert_eq!(imported.provider_type, jenkins.provider_type);
1846 assert_eq!(imported.refresh_interval, jenkins.refresh_interval);
1847 assert_eq!(imported.config, jenkins.config);
1848 assert_eq!(target.bus.count("providers-changed"), 1);
1849 }
1850
1851 #[tokio::test]
1852 async fn test_import_providers_toml_rejects_unknown_type_before_adding() {
1853 let h = Harness::with_real_plugins().await;
1854 let content = r#"
1855[providers.github-main]
1856type = "github"
1857
1858[providers.mystery]
1859type = "travis"
1860"#;
1861
1862 let err = h
1863 .provider_service
1864 .import_providers_toml(content)
1865 .await
1866 .unwrap_err();
1867
1868 assert!(matches!(err, DomainError::InvalidProviderType(_)), "{err}");
1869 assert!(h.repository.list_providers().await.unwrap().is_empty());
1870 assert_eq!(h.bus.count("providers-changed"), 0);
1871 }
1872
1873 #[test]
1874 fn quota_pause_starts_under_ten_percent_until_reset() {
1875 let now = Utc::now();
1876 let reset_at = now + chrono::Duration::minutes(20);
1877 let pause = |remaining, reset_at| {
1878 quota_pause(
1879 &RateLimit {
1880 limit: 5000,
1881 remaining,
1882 reset_at,
1883 },
1884 now,
1885 )
1886 };
1887
1888 assert_eq!(pause(500, reset_at), None);
1889 assert_eq!(pause(499, reset_at), Some(reset_at));
1890 assert_eq!(pause(0, reset_at), Some(reset_at));
1891 assert_eq!(pause(0, now), None);
1892 assert_eq!(
1893 quota_pause(
1894 &RateLimit {
1895 limit: 0,
1896 remaining: 0,
1897 reset_at,
1898 },
1899 now,
1900 ),
1901 None
1902 );
1903 }
1904
1905 type StubGate = Option<(Arc<tokio::sync::Notify>, Arc<tokio::sync::Notify>)>;
1906
1907 async fn stub_github(
1908 remaining: u64, reset: i64, gate: StubGate,
1909 ) -> (String, Arc<Mutex<Vec<String>>>) {
1910 use tokio::io::{
1911 AsyncReadExt,
1912 AsyncWriteExt,
1913 };
1914
1915 let _ = rustls::crypto::ring::default_provider().install_default();
1916 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1917 let addr = listener.local_addr().unwrap();
1918 let requests: Arc<Mutex<Vec<String>>> = Arc::default();
1919 let log = requests.clone();
1920 tokio::spawn(async move {
1921 loop {
1922 let (mut stream, _) = listener.accept().await.unwrap();
1923 let mut request = Vec::new();
1924 let mut buffer = [0; 4096];
1925 while !request.windows(4).any(|window| window == b"\r\n\r\n") {
1926 let count = stream.read(&mut buffer).await.unwrap();
1927 if count == 0 {
1928 break;
1929 }
1930 request.extend_from_slice(&buffer[..count]);
1931 }
1932 let request = String::from_utf8(request).unwrap();
1933 let path = request.split_whitespace().nth(1).unwrap_or_default();
1934 let path = path.split('?').next().unwrap().to_string();
1935 let (status, body) = if path == "/api/v3/rate_limit" {
1936 if let Some((arrived, release)) = &gate {
1937 arrived.notify_one();
1938 release.notified().await;
1939 }
1940 let rate = serde_json::json!({
1941 "limit": 5000, "used": 5000 - remaining, "remaining": remaining, "reset": reset
1942 });
1943 let search = serde_json::json!({"limit": 30, "used": 0, "remaining": 30, "reset": reset});
1944 let body = serde_json::json!({
1945 "resources": {"core": rate, "search": search},
1946 "rate": rate
1947 });
1948 ("200 OK", body.to_string())
1949 } else {
1950 ("404 Not Found", r#"{"message":"Not Found"}"#.to_string())
1951 };
1952 log.lock().await.push(path);
1953 let response = format!(
1954 "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
1955 body.len()
1956 );
1957 let _ = stream.write_all(response.as_bytes()).await;
1958 }
1959 });
1960
1961 (format!("http://{addr}"), requests)
1962 }
1963
1964 #[tokio::test]
1965 async fn refresh_skips_a_github_provider_with_its_quota_almost_used_up() {
1966 let h = Harness::with_real_plugins().await;
1967 h.start().await;
1968 let reset = Utc::now().timestamp() + 1200;
1969 let (base_url, requests) = stub_github(0, reset, None).await;
1970 let id = h
1971 .repository
1972 .add_provider(&provider(
1973 "github",
1974 "github",
1975 "ghp_stub",
1976 30,
1977 &[("base_url", &base_url), ("selected_items", "acme/app")],
1978 ))
1979 .await
1980 .unwrap();
1981
1982 let first = h
1983 .pipeline_service
1984 .fetch_now(&[id], Duration::from_secs(5))
1985 .await;
1986 let again = h
1987 .pipeline_service
1988 .fetch_now(&[id], Duration::from_secs(5))
1989 .await;
1990 let summaries = h.provider_service.list_providers().await.unwrap();
1991
1992 assert_eq!(*requests.lock().await, ["/api/v3/rate_limit"]);
1993 assert_eq!(
1994 summaries[0].paused_until,
1995 DateTime::from_timestamp(reset, 0)
1996 );
1997 assert!(matches!(summaries[0].last_fetch_status, FetchStatus::Never));
1998 assert_eq!(first[0].state, crate::domain::RefreshState::Recent);
1999 assert_eq!(again[0].state, crate::domain::RefreshState::Recent);
2000 assert_eq!(again[0].kind, Some(crate::domain::FailureKind::RateLimited));
2001 assert_eq!(*requests.lock().await, ["/api/v3/rate_limit"]);
2002 }
2003
2004 #[tokio::test]
2005 async fn refresh_fetches_a_github_provider_with_quota_left() {
2006 let h = Harness::with_real_plugins().await;
2007 h.start().await;
2008 let (base_url, requests) = stub_github(4000, Utc::now().timestamp() + 1200, None).await;
2009 let id = h
2010 .repository
2011 .add_provider(&provider(
2012 "github",
2013 "github",
2014 "ghp_stub",
2015 30,
2016 &[("base_url", &base_url), ("selected_items", "acme/app")],
2017 ))
2018 .await
2019 .unwrap();
2020
2021 h.pipeline_service
2022 .fetch_now(&[id], Duration::from_secs(5))
2023 .await;
2024 let summaries = h.provider_service.list_providers().await.unwrap();
2025 let requests = requests.lock().await;
2026
2027 assert_eq!(requests[0], "/api/v3/rate_limit");
2028 assert!(requests
2029 .iter()
2030 .any(|path| path.starts_with("/api/v3/repos/acme/app")));
2031 assert_eq!(summaries[0].paused_until, None);
2032 }
2033
2034 #[tokio::test]
2035 async fn quota_answer_for_replaced_credentials_does_not_pause_the_new_ones() {
2036 let h = Harness::with_real_plugins().await;
2037 let reset = Utc::now().timestamp() + 1200;
2038 let arrived = Arc::new(tokio::sync::Notify::new());
2039 let release = Arc::new(tokio::sync::Notify::new());
2040 let (base_url, _) = stub_github(0, reset, Some((arrived.clone(), release.clone()))).await;
2041 let id = h
2042 .repository
2043 .add_provider(&provider(
2044 "github",
2045 "github",
2046 "ghp_old",
2047 30,
2048 &[("base_url", &base_url), ("selected_items", "acme/app")],
2049 ))
2050 .await
2051 .unwrap();
2052 let service = Arc::clone(&h.provider_service);
2053 let lease = service.acquire(id).await.unwrap();
2054
2055 let check = tokio::spawn({
2056 let service = Arc::clone(&service);
2057 let lease = lease.clone();
2058 async move { service.check_quota(&lease).await }
2059 });
2060 arrived.notified().await;
2061 let replacement =
2062 (service.provider_factory)(&h.repository.get_provider(id).await.unwrap()).unwrap();
2063 service.replace(id, replacement, 30).await;
2064 release.notify_one();
2065
2066 assert_eq!(check.await.unwrap(), None);
2067 assert_eq!(h.scheduler.paused_until(id, Utc::now()), None);
2068 }
2069
2070 fn updated(h: &Harness, id: i64) -> ProviderConfig {
2071 let mut config = h.config("renamed", 45);
2072 config.id = Some(id);
2073 config
2074 }
2075
2076 #[tokio::test]
2077 async fn an_update_during_a_blocked_fetch_supersedes_the_old_attempt() {
2078 let h = Harness::new().await;
2079 h.start().await;
2080 let old_gate = Gate::new();
2081 let (id, old) = h
2082 .add_stub(|id| {
2083 StubProvider::new(id)
2084 .gated(&old_gate)
2085 .returning(vec![pipeline(id, "old", PipelineStatus::Success)])
2086 })
2087 .await;
2088 let new = h.stubs.queue(StubProvider::new(id).returning(vec![pipeline(
2089 id,
2090 "new",
2091 PipelineStatus::Failed,
2092 )]));
2093 let blocked = h.spawn_fetch(id).await;
2094 old_gate.arrived().await;
2095 let before = h.provider_service.acquire(id).await.unwrap().generation;
2096
2097 h.provider_service
2098 .update_provider(id, updated(&h, id))
2099 .await
2100 .unwrap();
2101 let updated_at = h
2102 .bus
2103 .names()
2104 .iter()
2105 .position(|name| *name == "provider-updated")
2106 .unwrap();
2107 h.bus
2108 .wait_for("the fetch queued by the update", |events| {
2109 events[updated_at..]
2110 .iter()
2111 .any(|event| matches!(event, CoreEvent::PipelinesUpdated { .. }))
2112 })
2113 .await;
2114 old_gate.open();
2115 let old_result = blocked.await.unwrap();
2116
2117 assert!(matches!(old_result, FetchResult::Superseded));
2118 assert_eq!(h.cached_names(id).await, ["new"]);
2119 let published: Vec<_> = h.bus.events()[updated_at..]
2120 .iter()
2121 .filter_map(|event| match event {
2122 CoreEvent::PipelinesUpdated { pipelines, .. } => {
2123 Some(pipelines.iter().map(|p| p.name.clone()).collect::<Vec<_>>())
2124 }
2125 _ => None,
2126 })
2127 .collect();
2128 assert_eq!(published, vec![vec!["new".to_string()]]);
2129 let view = h.scheduler.view(id, Instant::now()).unwrap();
2130 assert_eq!(
2131 (view.failures, view.last_failure, view.in_flight),
2132 (0, None, false)
2133 );
2134 assert_ne!(
2135 h.provider_service.acquire(id).await.unwrap().generation,
2136 before
2137 );
2138 assert_eq!((old.fetch_count(), new.fetch_count()), (1, 1));
2139 }
2140
2141 #[tokio::test]
2142 async fn an_update_resets_the_fetch_status_and_the_schedule() {
2143 let h = Harness::new().await;
2144 let (id, _) = h
2145 .add_stub(|id| StubProvider::new(id).failing(DomainError::NetworkError("down".into())))
2146 .await;
2147 h.fetch_directly(id).await;
2148 h.health_inputs();
2149 assert_eq!(h.scheduler.view(id, Instant::now()).unwrap().failures, 1);
2150
2151 h.provider_service
2152 .update_provider(id, updated(&h, id))
2153 .await
2154 .unwrap();
2155 let inputs = h.health_inputs();
2156
2157 let event = h
2158 .bus
2159 .events()
2160 .into_iter()
2161 .find_map(|event| match event {
2162 CoreEvent::ProviderUpdated { provider, .. } => Some(provider),
2163 _ => None,
2164 })
2165 .unwrap();
2166 assert!(matches!(event.last_fetch_status, FetchStatus::Never));
2167 assert_eq!(event.last_fetch_error, None);
2168 assert_eq!(event.last_fetch_error_kind, None);
2169 assert_eq!(event.health.next_attempt_at, None);
2170 assert_eq!(event.name, "renamed");
2171 assert_eq!(h.scheduler.view(id, Instant::now()).unwrap().failures, 0);
2172 let lease = h.provider_service.acquire(id).await.unwrap();
2173 assert!(matches!(
2174 inputs.as_slice(),
2175 [HealthInput::ProvidersChanged { changed }]
2176 if changed.len() == 1 && changed[0].generation == Some(lease.generation)
2177 ));
2178 }
2179
2180 #[tokio::test]
2181 async fn an_old_failure_finishing_after_an_update_leaves_the_new_schedule_clean() {
2182 let h = Harness::new().await;
2183 h.start().await;
2184 let old_gate = Gate::new();
2185 let new_gate = Gate::new();
2186 let (id, _) = h
2187 .add_stub(|id| {
2188 StubProvider::new(id)
2189 .gated(&old_gate)
2190 .failing(DomainError::NetworkError("down".into()))
2191 })
2192 .await;
2193 h.stubs.queue(StubProvider::new(id).gated(&new_gate));
2194 let blocked = h.spawn_fetch(id).await;
2195 old_gate.arrived().await;
2196
2197 h.provider_service
2198 .update_provider(id, updated(&h, id))
2199 .await
2200 .unwrap();
2201 new_gate.arrived().await;
2202 old_gate.open();
2203 let old_result = blocked.await.unwrap();
2204
2205 assert!(matches!(old_result, FetchResult::Superseded));
2206 let view = h.scheduler.view(id, Instant::now()).unwrap();
2207 assert_eq!(
2208 (view.failures, view.last_failure, view.in_flight),
2209 (0, None, true)
2210 );
2211 let fetch = h.repository.get_provider_fetch_status(id).await.unwrap();
2212 assert_eq!((fetch.status.as_str(), fetch.error), ("never", None));
2213 new_gate.open();
2214 h.bus
2215 .wait_for("the new fetch", |events| {
2216 events
2217 .iter()
2218 .any(|event| matches!(event, CoreEvent::PipelinesUpdated { .. }))
2219 })
2220 .await;
2221 assert_eq!(h.scheduler.view(id, Instant::now()).unwrap().failures, 0);
2222 }
2223
2224 #[tokio::test]
2225 async fn a_removal_during_a_blocked_fetch_leaves_nothing_behind() {
2226 let h = Harness::new().await;
2227 let gate = Gate::new();
2228 let (id, _) = h.add_stub(|id| StubProvider::new(id).gated(&gate)).await;
2229 let blocked = h.spawn_fetch(id).await;
2230 gate.arrived().await;
2231 h.health_inputs();
2232
2233 h.provider_service.remove_provider(id).await.unwrap();
2234 let removed_at = h.bus.events().len();
2235 gate.open();
2236 let result = blocked.await.unwrap();
2237 let inputs = h.health_inputs();
2238
2239 assert!(matches!(result, FetchResult::Superseded));
2240 assert!(h.repository.get_provider(id).await.is_err());
2241 assert!(h.repository.get_provider_fetch_status(id).await.is_err());
2242 assert!(h.cached_names(id).await.is_empty());
2243 assert!(h.scheduler.view(id, Instant::now()).is_none());
2244 assert!(h.bus.names()[removed_at..].is_empty());
2245 assert_eq!(h.bus.names()[removed_at - 1], "provider-removed");
2246 assert!(matches!(
2247 inputs.as_slice(),
2248 [HealthInput::ProvidersChanged { changed }]
2249 if changed[0].id == id && changed[0].generation.is_none()
2250 ));
2251 }
2252
2253 #[tokio::test]
2254 async fn concurrent_acquires_share_one_provider() {
2255 let h = Harness::new().await;
2256 let id = h.add_provider("shared").await;
2257
2258 let leases =
2259 futures::future::join_all((0..8).map(|_| h.provider_service.acquire(id))).await;
2260
2261 let first = leases[0].as_ref().unwrap();
2262 for lease in &leases {
2263 let lease = lease.as_ref().unwrap();
2264 assert!(Arc::ptr_eq(&lease.provider, &first.provider));
2265 assert_eq!(lease.generation, first.generation);
2266 }
2267 }
2268
2269 #[tokio::test]
2270 async fn a_lazy_acquire_racing_a_removal_does_not_bring_the_provider_back() {
2271 let h = Harness::new().await;
2272 let id = h.add_provider("racing").await;
2273 let lock = h.provider_service.commit_lock(id);
2274 let held = lock.lock().await;
2275 let removal = tokio::spawn({
2276 let service = Arc::clone(&h.provider_service);
2277 async move { service.remove_provider(id).await }
2278 });
2279 tokio::time::sleep(Duration::from_millis(100)).await;
2280 let acquire = tokio::spawn({
2281 let service = Arc::clone(&h.provider_service);
2282 async move { service.acquire(id).await }
2283 });
2284 tokio::time::sleep(Duration::from_millis(100)).await;
2285
2286 drop(held);
2287 removal.await.unwrap().unwrap();
2288 let acquired = acquire.await.unwrap();
2289
2290 assert!(matches!(acquired, Err(DomainError::ProviderNotFound(_))));
2291 assert!(h.provider_service.current(id).await.is_none());
2292 assert!(h.scheduler.view(id, Instant::now()).is_none());
2293 }
2294
2295 #[tokio::test]
2296 async fn changing_the_refresh_interval_keeps_the_generation_and_reschedules() {
2297 let h = Harness::new().await;
2298 h.start().await;
2299 let (id, _) = h.add_stub(StubProvider::new).await;
2300 h.pipeline_service
2301 .fetch_now(&[id], Duration::from_secs(5))
2302 .await;
2303 let before = h.provider_service.acquire(id).await.unwrap().generation;
2304 assert!(
2305 h.scheduler
2306 .view(id, Instant::now())
2307 .unwrap()
2308 .next_attempt_in
2309 > Duration::from_secs(20)
2310 );
2311
2312 h.provider_service
2313 .update_provider_refresh_interval(id, 5)
2314 .await
2315 .unwrap();
2316
2317 let after = h.provider_service.acquire(id).await.unwrap().generation;
2318 assert_eq!(before, after);
2319 assert!(
2320 h.scheduler
2321 .view(id, Instant::now())
2322 .unwrap()
2323 .next_attempt_in
2324 <= Duration::from_secs(5)
2325 );
2326 assert_eq!(
2327 h.repository
2328 .get_provider(id)
2329 .await
2330 .unwrap()
2331 .refresh_interval,
2332 5
2333 );
2334 }
2335
2336 #[tokio::test]
2337 async fn health_hears_about_every_provider_change() {
2338 let h = Harness::new().await;
2339
2340 let id = h
2341 .provider_service
2342 .add_provider(h.config("argo", 30))
2343 .await
2344 .unwrap();
2345 let added = h.health_inputs();
2346 let added_generation = h.provider_service.acquire(id).await.unwrap().generation;
2347 h.provider_service
2348 .update_provider(id, updated(&h, id))
2349 .await
2350 .unwrap();
2351 let updated_inputs = h.health_inputs();
2352 h.provider_service.load_all_providers().await.unwrap();
2353 let reloaded = h.health_inputs();
2354 let reloaded_generation = h.provider_service.acquire(id).await.unwrap().generation;
2355 h.provider_service.remove_provider(id).await.unwrap();
2356 let removed = h.health_inputs();
2357
2358 let generations = |inputs: &[HealthInput]| match inputs {
2359 [HealthInput::ProvidersChanged { changed }] => changed
2360 .iter()
2361 .map(|input| (input.id, input.generation))
2362 .collect::<Vec<_>>(),
2363 other => panic!("unexpected health inputs: {other:?}"),
2364 };
2365 assert_eq!(generations(&added), [(id, Some(added_generation))]);
2366 let updated_generation = generations(&updated_inputs)[0].1;
2367 assert!(updated_generation.is_some());
2368 assert_ne!(updated_generation, Some(added_generation));
2369 assert_eq!(generations(&reloaded), [(id, Some(reloaded_generation))]);
2370 assert_ne!(Some(reloaded_generation), updated_generation);
2371 assert_eq!(generations(&removed), [(id, None)]);
2372 assert_eq!(h.bus.count("provider-added"), 1);
2373 assert_eq!(h.bus.count("provider-updated"), 1);
2374 assert_eq!(h.bus.count("provider-removed"), 1);
2375 }
2376
2377 #[tokio::test]
2378 async fn workflow_parameters_are_cached_for_ten_minutes() {
2379 let h = Harness::new().await;
2380 let (id, stub) = h.add_stub(StubProvider::new).await;
2381 let workflow = format!("github__{id}__web");
2382
2383 let first = h
2384 .provider_service
2385 .get_workflow_parameters(id, &workflow)
2386 .await
2387 .unwrap();
2388 h.provider_service
2389 .get_workflow_parameters(id, &workflow)
2390 .await
2391 .unwrap();
2392 let cached_fetches = stub
2393 .parameter_fetches
2394 .load(std::sync::atomic::Ordering::SeqCst);
2395 sqlx::query(
2396 "UPDATE workflow_parameters_cache SET cached_at = datetime('now', '-11 minutes')",
2397 )
2398 .execute(&h.pool)
2399 .await
2400 .unwrap();
2401 h.provider_service
2402 .get_workflow_parameters(id, &workflow)
2403 .await
2404 .unwrap();
2405
2406 assert_eq!(first.len(), 1);
2407 assert_eq!(cached_fetches, 1);
2408 assert_eq!(
2409 stub.parameter_fetches
2410 .load(std::sync::atomic::Ordering::SeqCst),
2411 2
2412 );
2413 }
2414
2415 #[tokio::test]
2416 async fn workflow_parameter_failures_keep_their_kind_and_leave_no_lock_behind() {
2417 let h = Harness::new().await;
2418 let (id, stub) = h.add_stub(StubProvider::new).await;
2419 *stub.parameters.lock().unwrap() = Err(DomainError::RateLimited {
2420 message: "slow".to_string(),
2421 retry_after: None,
2422 });
2423
2424 let failed = h
2425 .provider_service
2426 .get_workflow_parameters(id, "github__1__web")
2427 .await;
2428 let missing = h
2429 .provider_service
2430 .get_workflow_parameters(4242, "github__4242__web")
2431 .await;
2432
2433 assert!(matches!(failed, Err(DomainError::RateLimited { .. })));
2434 assert!(matches!(missing, Err(DomainError::ProviderNotFound(_))));
2435 assert!(h.provider_service.parameter_fetches.is_empty());
2436 }
2437
2438 #[tokio::test]
2439 async fn fetching_workflow_parameters_does_not_block_the_provider_map() {
2440 let h = Harness::new().await;
2441 let gate = Gate::new();
2442 let (id, _) = h
2443 .add_stub(|id| StubProvider::new(id).parameter_gated(&gate))
2444 .await;
2445 let other = h.add_provider("other").await;
2446 let fetch = tokio::spawn({
2447 let service = Arc::clone(&h.provider_service);
2448 async move { service.get_workflow_parameters(id, "github__1__web").await }
2449 });
2450 gate.arrived().await;
2451
2452 let acquired =
2453 tokio::time::timeout(Duration::from_secs(2), h.provider_service.acquire(other)).await;
2454 gate.open();
2455
2456 assert!(acquired.is_ok());
2457 fetch.await.unwrap().unwrap();
2458 }
2459
2460 #[tokio::test]
2461 async fn workflow_parameters_fetched_across_a_removal_are_not_cached() {
2462 let h = Harness::new().await;
2463 let gate = Gate::new();
2464 let (id, _) = h
2465 .add_stub(|id| StubProvider::new(id).parameter_gated(&gate))
2466 .await;
2467 let workflow = format!("github__{id}__web");
2468 let fetch = tokio::spawn({
2469 let service = Arc::clone(&h.provider_service);
2470 let workflow = workflow.clone();
2471 async move { service.get_workflow_parameters(id, &workflow).await }
2472 });
2473 gate.arrived().await;
2474
2475 h.provider_service.remove_provider(id).await.unwrap();
2476 gate.open();
2477
2478 assert_eq!(fetch.await.unwrap().unwrap().len(), 1);
2479 assert!(h
2480 .repository
2481 .get_cached_workflow_parameters(&workflow)
2482 .await
2483 .unwrap()
2484 .is_none());
2485 }
2486
2487 #[tokio::test]
2488 async fn field_options_work_when_the_plugin_rejects_the_partial_form() {
2489 let h = Harness::with_real_plugins().await;
2490 let kubeconfig = h._temp_dir.path().join("kubeconfig");
2491 std::fs::write(
2492 &kubeconfig,
2493 "apiVersion: v1\nkind: Config\nclusters:\n- name: c\n cluster: {server: 'https://127.0.0.1:6443'}\nusers:\n- name: u\n user: {token: t}\ncontexts:\n- name: ctx-b\n context: {cluster: c, user: u}\n- name: ctx-a\n context: {cluster: c, user: u}\ncurrent-context: ctx-a\n",
2494 )
2495 .unwrap();
2496 let config = HashMap::from([
2497 ("kubeconfig".to_string(), kubeconfig.display().to_string()),
2498 ("namespaces".to_string(), "Not_A_Namespace".to_string()),
2499 ]);
2500 assert!(h
2501 .provider_service
2502 .init_plugin("fluxcd", config.clone(), 0)
2503 .await
2504 .is_err());
2505
2506 let contexts = h
2507 .provider_service
2508 .field_options("fluxcd", "context", config)
2509 .await
2510 .unwrap();
2511
2512 assert_eq!(contexts, ["ctx-a", "ctx-b"]);
2513 }
2514
2515 #[tokio::test]
2516 async fn failed_field_options_report_why_the_plugin_could_not_initialize() {
2517 let h = Harness::with_real_plugins().await;
2518 let config = HashMap::from([
2519 (
2520 "kubeconfig".to_string(),
2521 h._temp_dir.path().join("missing").display().to_string(),
2522 ),
2523 ("namespaces".to_string(), "Not_A_Namespace".to_string()),
2524 ]);
2525
2526 let rejected = h
2527 .provider_service
2528 .field_options("fluxcd", "namespaces", config)
2529 .await;
2530
2531 assert!(
2532 matches!(rejected, Err(DomainError::InvalidConfig(message)) if message.contains("initialize"))
2533 );
2534 }
2535
2536 fn serve_tls(stream: std::net::TcpStream, config: Arc<rustls::ServerConfig>) {
2537 let Ok(connection) = rustls::ServerConnection::new(config) else {
2538 return;
2539 };
2540 let mut tls = rustls::StreamOwned::new(connection, stream);
2541 let mut request = Vec::new();
2542 let mut buffer = [0; 4096];
2543 while !request.windows(4).any(|window| window == b"\r\n\r\n") {
2544 match tls.read(&mut buffer) {
2545 Ok(0) | Err(_) => return,
2546 Ok(count) => request.extend_from_slice(&buffer[..count]),
2547 }
2548 }
2549 let body = r#"{"items":[]}"#;
2550 let _ = write!(
2551 tls,
2552 "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
2553 body.len()
2554 );
2555 let _ = tls.flush();
2556 tls.conn.send_close_notify();
2557 let _ = tls.flush();
2558 }
2559
2560 fn self_signed_argocd() -> String {
2561 let _ = rustls::crypto::ring::default_provider().install_default();
2562 let cert = rustls::pki_types::CertificateDer::from(
2563 include_bytes!("../../../tests/fixtures/self_signed_cert.der").to_vec(),
2564 );
2565 let key = rustls::pki_types::PrivateKeyDer::Pkcs8(
2566 include_bytes!("../../../tests/fixtures/self_signed_key.der")
2567 .to_vec()
2568 .into(),
2569 );
2570 let config = Arc::new(
2571 rustls::ServerConfig::builder()
2572 .with_no_client_auth()
2573 .with_single_cert(vec![cert], key)
2574 .unwrap(),
2575 );
2576 let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
2577 let addr = listener.local_addr().unwrap();
2578 std::thread::spawn(move || {
2579 for stream in listener.incoming().flatten() {
2580 let config = Arc::clone(&config);
2581 std::thread::spawn(move || serve_tls(stream, config));
2582 }
2583 });
2584
2585 format!("https://{addr}")
2586 }
2587
2588 fn argocd_settings(url: &str, insecure: bool) -> HashMap<String, String> {
2589 HashMap::from([
2590 ("server_url".to_string(), url.to_string()),
2591 ("token".to_string(), "token".to_string()),
2592 ("insecure".to_string(), insecure.to_string()),
2593 ])
2594 }
2595
2596 #[tokio::test]
2597 async fn argocd_insecure_tls_is_honored_by_add_validate_and_preview() {
2598 let h = Harness::with_real_plugins().await;
2599 let service = &h.provider_service;
2600 let url = self_signed_argocd();
2601 let page = Some(pipedash_plugin_api::PaginationParams {
2602 page: 1,
2603 page_size: 10,
2604 });
2605 let add = |insecure| {
2606 let mut config = h.config("argo", 30);
2607 config.config = argocd_settings(&url, insecure);
2608 config.config.remove("token");
2609 service.add_provider(config)
2610 };
2611
2612 let id = add(true).await.unwrap();
2613 let insecure = service
2614 .init_plugin("argocd", argocd_settings(&url, true), 0)
2615 .await
2616 .unwrap();
2617 let secure = service
2618 .init_plugin("argocd", argocd_settings(&url, false), 0)
2619 .await
2620 .unwrap();
2621 let rejected = add(false).await.unwrap_err();
2622
2623 assert!(id > 0);
2624 assert!(insecure.validate_credentials().await.unwrap());
2625 assert!(insecure
2626 .fetch_available_pipelines_filtered(None, None, page.clone())
2627 .await
2628 .unwrap()
2629 .items
2630 .is_empty());
2631 assert!(matches!(
2632 secure.validate_credentials().await,
2633 Err(PluginError::Tls(_))
2634 ));
2635 assert!(matches!(
2636 secure
2637 .fetch_available_pipelines_filtered(None, None, page)
2638 .await,
2639 Err(PluginError::Tls(_))
2640 ));
2641 assert!(matches!(rejected, DomainError::InvalidConfig(message) if message.contains("TLS")));
2642 }
2643
2644 fn config_file(h: &Harness, providers: &[(&str, &str, i64)]) {
2645 let mut content = String::new();
2646 for (key, name, refresh_interval) in providers {
2647 content.push_str(&format!(
2648 "[providers.{key}]\nname = \"{name}\"\ntype = \"argocd\"\ntoken = \"token-{key}\"\nrefresh_interval = {refresh_interval}\n\n[providers.{key}.config]\nserver_url = \"{}\"\n\n",
2649 h.argocd_url
2650 ));
2651 }
2652 std::fs::write(h.config_state.config_path(), content).unwrap();
2653 }
2654
2655 async fn provider_id(h: &Harness, key: &str) -> Option<i64> {
2656 h.repository
2657 .config_backend()
2658 .list_providers()
2659 .await
2660 .unwrap()
2661 .into_iter()
2662 .find(|p| p.name == key)
2663 .and_then(|p| p.id)
2664 }
2665
2666 #[tokio::test]
2667 async fn removal_preserves_the_provider_and_toml_when_token_deletion_fails() {
2668 let h = Harness::new().await;
2669 config_file(&h, &[("ci-a", "CI A", 30)]);
2670 h.provider_service
2671 .apply_config_file(&h.config_state)
2672 .await
2673 .unwrap();
2674 let id = provider_id(&h, "ci-a").await.unwrap();
2675 h.token_store.fail_next_delete();
2676
2677 assert!(h.provider_service.remove_provider(id).await.is_err());
2678 assert!(h.repository.get_provider(id).await.is_ok());
2679 assert!(h
2680 .config_state
2681 .load_providers()
2682 .unwrap()
2683 .contains_key("ci-a"));
2684
2685 h.provider_service.remove_provider(id).await.unwrap();
2686 h.config_state.reload().await.unwrap();
2687 h.config_state
2688 .sync_providers_from_toml(h.repository.config_backend().as_ref(), false)
2689 .await
2690 .unwrap();
2691 assert!(h.repository.list_providers().await.unwrap().is_empty());
2692 assert!(h.token_store.get_all_tokens().await.unwrap().is_empty());
2693 }
2694
2695 #[tokio::test]
2696 async fn removal_awaits_token_cleanup_before_deleting_the_provider() {
2697 let h = Harness::new().await;
2698 let id = h.add_provider("await-delete").await;
2699 let gate = Gate::new();
2700 h.token_store.gate_delete(Arc::clone(&gate));
2701 let remove = tokio::spawn({
2702 let service = Arc::clone(&h.provider_service);
2703 async move { service.remove_provider(id).await }
2704 });
2705 gate.arrived().await;
2706
2707 assert!(!remove.is_finished());
2708 assert!(h.repository.get_provider(id).await.is_ok());
2709
2710 gate.open();
2711 remove.await.unwrap().unwrap();
2712 assert!(h.repository.get_provider(id).await.is_err());
2713 assert!(h.token_store.get_all_tokens().await.unwrap().is_empty());
2714 }
2715
2716 #[tokio::test]
2717 async fn failed_config_deletion_keeps_the_provider_available_for_retry() {
2718 let h = Harness::new().await;
2719 config_file(&h, &[("ci-a", "CI A", 30)]);
2720 h.provider_service
2721 .apply_config_file(&h.config_state)
2722 .await
2723 .unwrap();
2724 let id = provider_id(&h, "ci-a").await.unwrap();
2725 let path = h.config_state.config_path();
2726 std::fs::remove_file(path).unwrap();
2727 std::fs::create_dir(path).unwrap();
2728
2729 assert!(h.provider_service.remove_provider(id).await.is_err());
2730 assert!(h.repository.get_provider(id).await.is_ok());
2731
2732 std::fs::remove_dir(path).unwrap();
2733 config_file(&h, &[("ci-a", "CI A", 30)]);
2734 h.provider_service.remove_provider(id).await.unwrap();
2735 assert!(h.repository.get_provider(id).await.is_err());
2736 assert!(h.config_state.load_providers().unwrap().is_empty());
2737 }
2738
2739 #[tokio::test]
2740 async fn renaming_a_toml_provider_keeps_the_file_key_for_removal() {
2741 let h = Harness::new().await;
2742 config_file(&h, &[("ci-a", "CI A", 30)]);
2743 h.provider_service
2744 .apply_config_file(&h.config_state)
2745 .await
2746 .unwrap();
2747 let id = provider_id(&h, "ci-a").await.unwrap();
2748 let mut renamed = h.repository.get_provider(id).await.unwrap();
2749 renamed.name = "Renamed CI".to_string();
2750
2751 h.provider_service
2752 .update_provider(id, renamed)
2753 .await
2754 .unwrap();
2755 let saved = h.repository.get_provider(id).await.unwrap();
2756 assert_eq!(
2757 (saved.name.as_str(), saved.display_name()),
2758 ("ci-a", "Renamed CI")
2759 );
2760
2761 h.provider_service.remove_provider(id).await.unwrap();
2762 h.config_state.reload().await.unwrap();
2763 h.config_state
2764 .sync_providers_from_toml(h.repository.config_backend().as_ref(), false)
2765 .await
2766 .unwrap();
2767 assert!(h.repository.list_providers().await.unwrap().is_empty());
2768 }
2769
2770 #[tokio::test]
2771 async fn test_apply_config_file_adds_and_updates_changed_blocks_only() {
2772 let h = Harness::new().await;
2773 config_file(&h, &[("ci-a", "CI A", 30), ("ci-b", "CI B", 30)]);
2774
2775 let first = h
2776 .provider_service
2777 .apply_config_file(&h.config_state)
2778 .await
2779 .unwrap();
2780
2781 assert_eq!(first.added, ["ci-a", "ci-b"]);
2782 let a = provider_id(&h, "ci-a").await.unwrap();
2783 let b = provider_id(&h, "ci-b").await.unwrap();
2784 let loaded = h.provider_service.generation_of(a).await.unwrap();
2785 assert!(h.provider_service.generation_of(b).await.is_some());
2786 assert!(h
2787 .bus
2788 .events()
2789 .iter()
2790 .any(|e| matches!(e, CoreEvent::ProvidersChanged)));
2791
2792 h.provider_service
2793 .update_provider_refresh_interval(b, 120)
2794 .await
2795 .unwrap();
2796 config_file(
2797 &h,
2798 &[
2799 ("ci-a", "CI Renamed", 30),
2800 ("ci-b", "CI B", 30),
2801 ("ci-c", "CI C", 60),
2802 ],
2803 );
2804
2805 let second = h
2806 .provider_service
2807 .apply_config_file(&h.config_state)
2808 .await
2809 .unwrap();
2810
2811 assert_eq!(second.added, ["ci-c"]);
2812 assert_eq!(second.updated, ["ci-a"]);
2813 let c = provider_id(&h, "ci-c").await.unwrap();
2814 assert_eq!(
2815 h.repository.get_provider(c).await.unwrap().token,
2816 "token-ci-c"
2817 );
2818 assert!(h.provider_service.generation_of(c).await.is_some());
2819 assert_eq!(
2820 h.repository.get_provider(a).await.unwrap().display_name(),
2821 "CI Renamed"
2822 );
2823 assert_ne!(h.provider_service.generation_of(a).await, Some(loaded));
2824 assert_eq!(
2825 h.repository.get_provider(b).await.unwrap().refresh_interval,
2826 120
2827 );
2828
2829 let unchanged = h
2830 .provider_service
2831 .apply_config_file(&h.config_state)
2832 .await
2833 .unwrap();
2834 assert!(unchanged.added.is_empty() && unchanged.updated.is_empty());
2835 }
2836
2837 #[tokio::test]
2838 async fn test_apply_config_file_rejects_an_invalid_file_and_keeps_providers() {
2839 let h = Harness::new().await;
2840 config_file(&h, &[("ci-a", "CI A", 30)]);
2841 h.provider_service
2842 .apply_config_file(&h.config_state)
2843 .await
2844 .unwrap();
2845 let a = provider_id(&h, "ci-a").await.unwrap();
2846
2847 let write = |content: &str| std::fs::write(h.config_state.config_path(), content).unwrap();
2848
2849 write("[providers.ci-a]\nname = \"CI Broken\"\ntype = \"\"\n");
2850 let empty = h.provider_service.apply_config_file(&h.config_state).await;
2851 write("[providers.ci-a]\nname = \"CI Broken\"\ntype = \"circleci\"\n");
2852 let unknown = h.provider_service.apply_config_file(&h.config_state).await;
2853
2854 assert!(matches!(empty, Err(DomainError::InvalidConfig(_))));
2855 assert!(matches!(unknown, Err(DomainError::InvalidProviderType(_))));
2856
2857 assert_eq!(
2858 h.repository.get_provider(a).await.unwrap().display_name(),
2859 "CI A"
2860 );
2861 assert_eq!(
2862 h.config_state.get().await.providers["ci-a"].provider_type,
2863 "argocd"
2864 );
2865 }
2866
2867 async fn execute(h: &Harness, sql: &'static str) {
2868 sqlx::query(sql).execute(&h.pool).await.unwrap();
2869 }
2870
2871 #[tokio::test]
2872 async fn test_apply_config_file_retry_replaces_providers_saved_by_a_failed_apply() {
2873 let h = Harness::new().await;
2874 config_file(&h, &[("ci-a", "CI A", 30), ("ci-b", "CI B", 30)]);
2875 h.provider_service
2876 .apply_config_file(&h.config_state)
2877 .await
2878 .unwrap();
2879 let a = provider_id(&h, "ci-a").await.unwrap();
2880 let b = provider_id(&h, "ci-b").await.unwrap();
2881 let loaded_a = h.provider_service.generation_of(a).await.unwrap();
2882 let loaded_b = h.provider_service.generation_of(b).await.unwrap();
2883
2884 execute(
2885 &h,
2886 "CREATE TRIGGER fail_ci_b BEFORE UPDATE ON providers WHEN NEW.name = 'ci-b' \
2887 BEGIN SELECT RAISE(ABORT, 'write failed'); END",
2888 )
2889 .await;
2890 config_file(&h, &[("ci-a", "CI A2", 30), ("ci-b", "CI B2", 30)]);
2891 assert!(h
2892 .provider_service
2893 .apply_config_file(&h.config_state)
2894 .await
2895 .is_err());
2896 assert_eq!(
2897 h.repository.get_provider(a).await.unwrap().display_name(),
2898 "CI A2"
2899 );
2900
2901 execute(&h, "DROP TRIGGER fail_ci_b").await;
2902 h.provider_service
2903 .apply_config_file(&h.config_state)
2904 .await
2905 .unwrap();
2906
2907 assert_eq!(
2908 h.repository.get_provider(b).await.unwrap().display_name(),
2909 "CI B2"
2910 );
2911 assert_ne!(h.provider_service.generation_of(a).await, Some(loaded_a));
2912 assert_ne!(h.provider_service.generation_of(b).await, Some(loaded_b));
2913 assert_eq!(
2914 h.config_state.get().await.providers["ci-b"].name.as_deref(),
2915 Some("CI B2")
2916 );
2917 }
2918
2919 #[tokio::test]
2920 async fn test_apply_config_file_stores_an_empty_token_for_a_new_provider_without_one() {
2921 let h = Harness::new().await;
2922 config_file(&h, &[("ci-a", "CI A", 30)]);
2923 let path = h.config_state.config_path();
2924 let unset = std::fs::read_to_string(path)
2925 .unwrap()
2926 .replace("token-ci-a", "${PIPEDASH_TOKEN_UNSET_IN_TESTS:-}");
2927 std::fs::write(path, unset).unwrap();
2928
2929 h.provider_service
2930 .apply_config_file(&h.config_state)
2931 .await
2932 .unwrap();
2933
2934 let a = provider_id(&h, "ci-a").await.unwrap();
2935 assert_eq!(h.token_store.get_token(a).await.unwrap(), "");
2936 }
2937
2938 #[tokio::test]
2939 async fn test_apply_config_file_retries_a_token_the_store_failed_to_save() {
2940 let h = Harness::new().await;
2941 config_file(&h, &[("ci-a", "CI A", 30)]);
2942 h.provider_service
2943 .apply_config_file(&h.config_state)
2944 .await
2945 .unwrap();
2946 let a = provider_id(&h, "ci-a").await.unwrap();
2947 let path = h.config_state.config_path();
2948 let rotated = std::fs::read_to_string(path)
2949 .unwrap()
2950 .replace("token-ci-a", "rotated");
2951 std::fs::write(path, rotated).unwrap();
2952
2953 h.token_store.fail_next_store();
2954 assert!(h
2955 .provider_service
2956 .apply_config_file(&h.config_state)
2957 .await
2958 .is_err());
2959 assert_eq!(h.stubs.token_of(a).as_deref(), Some("token-ci-a"));
2960
2961 h.provider_service
2962 .apply_config_file(&h.config_state)
2963 .await
2964 .unwrap();
2965
2966 assert_eq!(h.stubs.token_of(a).as_deref(), Some("rotated"));
2967 assert_eq!(
2968 h.config_state.get().await.providers["ci-a"].token,
2969 "rotated"
2970 );
2971 }
2972
2973 #[tokio::test]
2974 async fn test_config_watcher_retries_a_failed_apply_without_a_new_edit() {
2975 let h = Harness::new().await;
2976 let watcher = crate::application::ConfigWatcher::new(
2977 Arc::clone(&h.provider_service),
2978 Arc::clone(&h.config_state),
2979 )
2980 .with_poll(Duration::from_millis(20));
2981 watcher.start().await;
2982
2983 execute(
2984 &h,
2985 "CREATE TRIGGER fail_insert BEFORE INSERT ON providers \
2986 BEGIN SELECT RAISE(ABORT, 'write failed'); END",
2987 )
2988 .await;
2989 config_file(&h, &[("ci-a", "CI A", 30)]);
2990 tokio::time::sleep(Duration::from_millis(200)).await;
2991 assert_eq!(provider_id(&h, "ci-a").await, None);
2992 execute(&h, "DROP TRIGGER fail_insert").await;
2993
2994 let deadline = std::time::Instant::now() + Duration::from_secs(10);
2995 let id = loop {
2996 if let Some(id) = provider_id(&h, "ci-a").await {
2997 break id;
2998 }
2999 assert!(
3000 std::time::Instant::now() < deadline,
3001 "the watcher never retried the file"
3002 );
3003 tokio::time::sleep(Duration::from_millis(20)).await;
3004 };
3005 watcher.stop().await;
3006
3007 assert!(h.provider_service.generation_of(id).await.is_some());
3008 }
3009
3010 fn sign_in(token_endpoint: String, expires_in: chrono::Duration) -> SignInCredential {
3011 SignInCredential {
3012 token: "bearer".into(),
3013 refresh_token: Some("r1".into()),
3014 expires_at: Some(Utc::now() + expires_in),
3015 token_endpoint,
3016 client_id: "cli".into(),
3017 kind: pipedash_plugin_api::SignInToken::AccessToken,
3018 account: Some("dev@example.com".into()),
3019 }
3020 }
3021
3022 async fn stored_sign_in(h: &Harness, id: i64) -> SignInCredential {
3023 SignInCredential::decode(&h.repository.get_provider(id).await.unwrap().token).unwrap()
3024 }
3025
3026 async fn recovery_provider(h: &Harness) -> (i64, Idp, FeedStub) {
3027 let idp = Idp::start().await;
3028 idp.reply(
3029 200,
3030 serde_json::json!({
3031 "access_token": "fresh-access",
3032 "id_token": jwt(serde_json::json!({"email": "new@example.com", "exp": Utc::now().timestamp() + 3600})),
3033 "refresh_token": "fresh-refresh",
3034 }),
3035 );
3036 let argocd = FeedStub::start().await;
3037 argocd.respond_json(serde_json::json!({
3038 "oidcConfig": {"issuer": idp.base_url, "clientID": "cli"},
3039 }));
3040 let mut config = h.config("sso", 30);
3041 config
3042 .config
3043 .insert("server_url".into(), argocd.base_url.clone());
3044 config.token = sign_in(idp.token_endpoint(), chrono::Duration::minutes(30))
3045 .encode()
3046 .unwrap();
3047 let id = h.repository.add_provider(&config).await.unwrap();
3048 (id, idp, argocd)
3049 }
3050
3051 async fn complete_login(h: &Harness, start: &SignInStart) {
3052 let authorize = reqwest::Url::parse(&start.authorize_url).unwrap();
3053 let query: HashMap<_, _> = authorize.query_pairs().into_owned().collect();
3054 let mut callback = reqwest::Url::parse(&query["redirect_uri"]).unwrap();
3055 callback
3056 .query_pairs_mut()
3057 .append_pair("code", "code")
3058 .append_pair("state", &query["state"]);
3059 assert!(matches!(
3060 h.provider_service
3061 .complete_sign_in(&start.id, callback.as_str())
3062 .await
3063 .unwrap(),
3064 SignInStatus::Completed { .. }
3065 ));
3066 }
3067
3068 #[tokio::test]
3069 async fn provider_sign_in_starts_without_renewing_the_expired_credentials() {
3070 let h = Harness::new().await;
3071 let (id, idp, _argocd) = recovery_provider(&h).await;
3072 let expired = sign_in(idp.token_endpoint(), chrono::Duration::minutes(-1));
3073 h.repository
3074 .store_token(id, &expired.encode().unwrap())
3075 .await
3076 .unwrap();
3077 idp.reply(400, serde_json::json!({"error": "invalid_grant"}));
3078
3079 let start = h.provider_service.start_provider_sign_in(id).await.unwrap();
3080
3081 assert!(idp.forms.lock().unwrap().is_empty());
3082 assert_eq!(stored_sign_in(&h, id).await, expired);
3083 h.provider_service.cancel_sign_in(&start.id).await;
3084 }
3085
3086 #[tokio::test]
3087 async fn reconnect_preserves_concurrent_configuration_and_queues_a_fetch() {
3088 let h = Harness::new().await;
3089 let (id, _idp, argocd) = recovery_provider(&h).await;
3090 h.stubs.queue(
3091 StubProvider::new(id)
3092 .failing(DomainError::ProviderAuthenticationFailed("expired".into())),
3093 );
3094 h.fetch_directly(id).await;
3095 let generation = h.provider_service.generation_of(id).await.unwrap();
3096 let start = h.provider_service.start_provider_sign_in(id).await.unwrap();
3097 complete_login(&h, &start).await;
3098 let settings = Gate::new();
3099 argocd.set_gate(Some(&settings));
3100 let fetch = Gate::new();
3101 h.stubs.queue(StubProvider::new(id).gated(&fetch));
3102 h.start().await;
3103 let reconnect = tokio::spawn({
3104 let service = Arc::clone(&h.provider_service);
3105 let flow = start.id.clone();
3106 async move { service.reconnect_provider(id, &flow).await }
3107 });
3108 settings.arrived().await;
3109 let lock = h.provider_service.commit_lock(id);
3110 let guard = lock.lock().await;
3111 let mut edited = h.repository.get_provider(id).await.unwrap();
3112 edited.name = "Renamed while signing in".into();
3113 edited.refresh_interval = 90;
3114 edited
3115 .config
3116 .insert("selected_items".into(), "new-app".into());
3117 h.repository.update_provider(id, &edited).await.unwrap();
3118 let edited = h.repository.get_provider(id).await.unwrap();
3119 drop(guard);
3120 settings.open();
3121 reconnect.await.unwrap().unwrap();
3122 fetch.arrived().await;
3123
3124 let saved = h.repository.get_provider(id).await.unwrap();
3125 assert_eq!(saved.name, edited.name);
3126 assert_eq!(saved.refresh_interval, edited.refresh_interval);
3127 assert_eq!(saved.config, edited.config);
3128 assert_eq!(saved.version, edited.version);
3129 assert_eq!(
3130 stored_sign_in(&h, id).await.account.as_deref(),
3131 Some("new@example.com")
3132 );
3133 assert_eq!(
3134 stored_sign_in(&h, id).await.refresh_token.as_deref(),
3135 Some("fresh-refresh")
3136 );
3137 assert_ne!(
3138 h.provider_service.generation_of(id).await.unwrap(),
3139 generation
3140 );
3141 let schedule = h.scheduler.view(id, Instant::now()).unwrap();
3142 assert_eq!(
3143 (schedule.failures, schedule.last_failure, schedule.in_flight),
3144 (0, None, true)
3145 );
3146 assert_eq!(h.bus.count("provider-updated"), 1);
3147 assert!(h.bus.events().iter().any(|event| matches!(
3148 event,
3149 CoreEvent::ProviderUpdated { provider, .. }
3150 if matches!(provider.last_fetch_status, FetchStatus::Never)
3151 && provider.last_fetch_error.is_none()
3152 && provider.recovery.is_none()
3153 )));
3154 assert_eq!(
3155 h.provider_service.sign_in_status(&start.id).await.unwrap(),
3156 SignInStatus::Expired
3157 );
3158 fetch.open();
3159 h.bus
3160 .wait_for("the reconnected provider fetch", |events| {
3161 events
3162 .iter()
3163 .any(|event| matches!(event, CoreEvent::PipelinesUpdated { .. }))
3164 })
3165 .await;
3166 }
3167
3168 #[tokio::test]
3169 async fn reconnect_retains_the_flow_and_runtime_when_token_persistence_fails() {
3170 let h = Harness::new().await;
3171 let (id, _idp, _argocd) = recovery_provider(&h).await;
3172 let original = h.repository.get_provider(id).await.unwrap();
3173 let lease = h.provider_service.acquire(id).await.unwrap();
3174 let start = h.provider_service.start_provider_sign_in(id).await.unwrap();
3175 complete_login(&h, &start).await;
3176 h.token_store.fail_next_store();
3177
3178 assert!(h
3179 .provider_service
3180 .reconnect_provider(id, &start.id)
3181 .await
3182 .is_err());
3183
3184 assert_eq!(
3185 h.repository.get_provider(id).await.unwrap().token,
3186 original.token
3187 );
3188 assert_eq!(
3189 h.provider_service.generation_of(id).await,
3190 Some(lease.generation)
3191 );
3192 assert_eq!(h.bus.count("provider-updated"), 0);
3193 assert!(matches!(
3194 h.provider_service.sign_in_status(&start.id).await.unwrap(),
3195 SignInStatus::Completed { .. }
3196 ));
3197 h.provider_service
3198 .reconnect_provider(id, &start.id)
3199 .await
3200 .unwrap();
3201 assert_eq!(h.bus.count("provider-updated"), 1);
3202 assert_eq!(
3203 h.provider_service.sign_in_status(&start.id).await.unwrap(),
3204 SignInStatus::Expired
3205 );
3206 }
3207
3208 #[tokio::test]
3209 async fn reconnect_rejects_other_providers_setup_flows_and_deleted_providers() {
3210 let h = Harness::new().await;
3211 let (id, _idp, _argocd) = recovery_provider(&h).await;
3212 let other = h.add_provider("other").await;
3213 let start = h.provider_service.start_provider_sign_in(id).await.unwrap();
3214 complete_login(&h, &start).await;
3215
3216 assert!(matches!(
3217 h.provider_service
3218 .reconnect_provider(other, &start.id)
3219 .await,
3220 Err(DomainError::InvalidConfig(_))
3221 ));
3222 let mut stolen = h.repository.get_provider(other).await.unwrap();
3223 stolen
3224 .config
3225 .insert(SIGN_IN_CONFIG_KEY.into(), start.id.clone());
3226 assert!(matches!(
3227 h.provider_service.update_provider(other, stolen).await,
3228 Err(DomainError::InvalidConfig(_))
3229 ));
3230 let config = h.repository.get_provider(id).await.unwrap();
3231 let setup = h
3232 .provider_service
3233 .start_sign_in(&config.provider_type, config.config)
3234 .await
3235 .unwrap();
3236 complete_login(&h, &setup).await;
3237 assert!(matches!(
3238 h.provider_service.reconnect_provider(id, &setup.id).await,
3239 Err(DomainError::InvalidConfig(_))
3240 ));
3241 h.provider_service.remove_provider(id).await.unwrap();
3242 assert!(matches!(
3243 h.provider_service.reconnect_provider(id, &start.id).await,
3244 Err(DomainError::ProviderNotFound(_))
3245 ));
3246 assert_eq!(h.bus.count("provider-updated"), 0);
3247 assert_eq!(
3248 h.repository.get_provider(other).await.unwrap().token,
3249 "token"
3250 );
3251 }
3252
3253 #[tokio::test]
3254 async fn deleting_during_reconnect_does_not_restore_the_provider() {
3255 let h = Harness::new().await;
3256 let (id, _idp, argocd) = recovery_provider(&h).await;
3257 let start = h.provider_service.start_provider_sign_in(id).await.unwrap();
3258 complete_login(&h, &start).await;
3259 let settings = Gate::new();
3260 argocd.set_gate(Some(&settings));
3261 let reconnect = tokio::spawn({
3262 let service = Arc::clone(&h.provider_service);
3263 async move { service.reconnect_provider(id, &start.id).await }
3264 });
3265 settings.arrived().await;
3266
3267 h.provider_service.remove_provider(id).await.unwrap();
3268 settings.open();
3269
3270 assert!(matches!(
3271 reconnect.await.unwrap(),
3272 Err(DomainError::ProviderNotFound(_))
3273 ));
3274 assert!(h.repository.get_provider(id).await.is_err());
3275 assert!(h.provider_service.generation_of(id).await.is_none());
3276 assert!(h.scheduler.view(id, Instant::now()).is_none());
3277 assert_eq!(h.bus.count("provider-updated"), 0);
3278 }
3279
3280 #[tokio::test]
3281 async fn reconnect_rejects_replaced_credentials_and_changed_sign_in_settings() {
3282 let h = Harness::new().await;
3283 let (id, _idp, argocd) = recovery_provider(&h).await;
3284 let original = h.repository.get_provider(id).await.unwrap();
3285 let start = h.provider_service.start_provider_sign_in(id).await.unwrap();
3286 complete_login(&h, &start).await;
3287 h.repository
3288 .store_token(id, "changed-by-another-client")
3289 .await
3290 .unwrap();
3291
3292 assert!(matches!(
3293 h.provider_service.reconnect_provider(id, &start.id).await,
3294 Err(DomainError::ConcurrentModification(_))
3295 ));
3296 assert_eq!(
3297 h.repository.get_provider(id).await.unwrap().token,
3298 "changed-by-another-client"
3299 );
3300 h.repository.store_token(id, &original.token).await.unwrap();
3301 argocd.respond_json(serde_json::json!({
3302 "oidcConfig": {"issuer": "https://different.example.com", "clientID": "different"},
3303 }));
3304 assert!(matches!(
3305 h.provider_service.reconnect_provider(id, &start.id).await,
3306 Err(DomainError::ConcurrentModification(_))
3307 ));
3308 assert_eq!(
3309 h.repository.get_provider(id).await.unwrap().token,
3310 original.token
3311 );
3312 assert_eq!(h.bus.count("provider-updated"), 0);
3313 }
3314
3315 #[tokio::test]
3316 async fn concurrent_reconnects_commit_a_completed_flow_only_once() {
3317 let h = Harness::new().await;
3318 let (id, _idp, _argocd) = recovery_provider(&h).await;
3319 let start = h.provider_service.start_provider_sign_in(id).await.unwrap();
3320 complete_login(&h, &start).await;
3321
3322 let (first, second) = tokio::join!(
3323 h.provider_service.reconnect_provider(id, &start.id),
3324 h.provider_service.reconnect_provider(id, &start.id),
3325 );
3326
3327 assert_ne!(first.is_ok(), second.is_ok());
3328 assert_eq!(h.bus.count("provider-updated"), 1);
3329 assert_eq!(
3330 h.provider_service.sign_in_status(&start.id).await.unwrap(),
3331 SignInStatus::Expired
3332 );
3333 }
3334
3335 #[tokio::test]
3336 async fn reconnect_requires_completion_and_cancel_discards_completed_recovery() {
3337 let h = Harness::new().await;
3338 let (id, _idp, _argocd) = recovery_provider(&h).await;
3339 let original = h.repository.get_provider(id).await.unwrap().token;
3340 let start = h.provider_service.start_provider_sign_in(id).await.unwrap();
3341
3342 assert!(matches!(
3343 h.provider_service.reconnect_provider(id, &start.id).await,
3344 Err(DomainError::InvalidConfig(_))
3345 ));
3346 complete_login(&h, &start).await;
3347 h.provider_service.cancel_sign_in(&start.id).await;
3348 assert_eq!(
3349 h.provider_service.sign_in_status(&start.id).await.unwrap(),
3350 SignInStatus::Expired
3351 );
3352 assert!(h
3353 .provider_service
3354 .reconnect_provider(id, &start.id)
3355 .await
3356 .is_err());
3357 assert_eq!(h.repository.get_provider(id).await.unwrap().token, original);
3358 assert_eq!(h.bus.count("provider-updated"), 0);
3359 }
3360
3361 #[tokio::test]
3362 async fn renew_is_scheduled_a_minute_before_the_sign_in_expires() {
3363 let h = Harness::with_real_plugins().await;
3364 let id = h.add_provider("sso").await;
3365 let credential = sign_in(
3366 "http://127.0.0.1:9/token".into(),
3367 chrono::Duration::minutes(30),
3368 );
3369 h.repository
3370 .store_token(id, &credential.encode().unwrap())
3371 .await
3372 .unwrap();
3373
3374 let lease = h.provider_service.acquire(id).await.unwrap();
3375
3376 assert_eq!(
3377 lease.provider.renew_at(),
3378 Some(credential.expires_at.unwrap() - chrono::Duration::seconds(60))
3379 );
3380 }
3381
3382 #[tokio::test]
3383 async fn renew_refreshes_and_stores_an_expired_sign_in() {
3384 let h = Harness::new().await;
3385 let idp = Idp::start().await;
3386 idp.reply(
3387 200,
3388 serde_json::json!({"access_token": "new", "refresh_token": "r2", "expires_in": 3600}),
3389 );
3390 let (id, _) = h.add_stub(StubProvider::new).await;
3391 let expired = sign_in(idp.token_endpoint(), chrono::Duration::minutes(-1));
3392 h.repository
3393 .store_token(id, &expired.encode().unwrap())
3394 .await
3395 .unwrap();
3396
3397 h.provider_service.acquire(id).await.unwrap();
3398
3399 let renewed = stored_sign_in(&h, id).await;
3400 assert_eq!(renewed.token, "new");
3401 assert_eq!(renewed.refresh_token.as_deref(), Some("r2"));
3402 assert_eq!(renewed.account.as_deref(), Some("dev@example.com"));
3403 }
3404
3405 #[tokio::test]
3406 async fn renew_with_a_revoked_sign_in_asks_to_sign_in_again() {
3407 let h = Harness::new().await;
3408 let idp = Idp::start().await;
3409 idp.reply(400, serde_json::json!({"error": "invalid_grant"}));
3410 let (id, _) = h.add_stub(StubProvider::new).await;
3411 let expired = sign_in(idp.token_endpoint(), chrono::Duration::minutes(-1));
3412 h.repository
3413 .store_token(id, &expired.encode().unwrap())
3414 .await
3415 .unwrap();
3416
3417 let error = h.provider_service.acquire(id).await.err().unwrap();
3418
3419 assert!(
3420 matches!(error, DomainError::ProviderAuthenticationFailed(m) if m == crate::domain::SIGN_IN_EXPIRED)
3421 );
3422 }
3423
3424 #[tokio::test]
3425 async fn stored_dex_refresh_rejections_ask_to_sign_in_again() {
3426 let h = Harness::new().await;
3427 let idp = Idp::start().await;
3428 let id = h.add_provider("sso").await;
3429 let expired = sign_in(idp.token_endpoint(), chrono::Duration::minutes(-1));
3430 h.repository
3431 .store_token(id, &expired.encode().unwrap())
3432 .await
3433 .unwrap();
3434
3435 for description in [
3436 "Refresh token is invalid or has already been claimed by another client.",
3437 "Refresh token expired.",
3438 ] {
3439 idp.reply(
3440 400,
3441 serde_json::json!({"error": "invalid_request", "error_description": description}),
3442 );
3443
3444 let error = h.provider_service.acquire(id).await.err().unwrap();
3445
3446 assert!(matches!(
3447 error,
3448 DomainError::ProviderAuthenticationFailed(_)
3449 ));
3450 assert_eq!(
3451 ProviderRecovery::for_failure(Some(error.failure_kind()), true),
3452 Some(ProviderRecovery::SignIn)
3453 );
3454 assert_eq!(stored_sign_in(&h, id).await, expired);
3455 }
3456 }
3457
3458 #[tokio::test]
3459 async fn unknown_dex_invalid_request_stays_an_api_error() {
3460 let h = Harness::new().await;
3461 let idp = Idp::start().await;
3462 idp.reply(
3463 400,
3464 serde_json::json!({"error": "invalid_request", "error_description": "Unknown request"}),
3465 );
3466 let id = h.add_provider("sso").await;
3467 let expired = sign_in(idp.token_endpoint(), chrono::Duration::minutes(-1));
3468 h.repository
3469 .store_token(id, &expired.encode().unwrap())
3470 .await
3471 .unwrap();
3472
3473 let error = h.provider_service.acquire(id).await.err().unwrap();
3474
3475 assert!(matches!(error, DomainError::ApiError(_)));
3476 }
3477
3478 #[tokio::test]
3479 async fn dex_invalid_request_server_errors_stay_transient() {
3480 let h = Harness::new().await;
3481 let idp = Idp::start().await;
3482 idp.reply(
3483 503,
3484 serde_json::json!({
3485 "error": "invalid_request",
3486 "error_description": "Refresh token expired.",
3487 }),
3488 );
3489 let id = h.add_provider("sso").await;
3490 let expired = sign_in(idp.token_endpoint(), chrono::Duration::minutes(-1));
3491 h.repository
3492 .store_token(id, &expired.encode().unwrap())
3493 .await
3494 .unwrap();
3495
3496 let error = h.provider_service.acquire(id).await.err().unwrap();
3497
3498 assert!(matches!(error, DomainError::ServiceUnavailable { .. }));
3499 }
3500
3501 #[tokio::test]
3502 async fn editing_a_signed_in_provider_keeps_its_session() {
3503 let h = Harness::with_real_plugins().await;
3504 let id = h.add_provider("sso").await;
3505 let credential = sign_in(
3506 "http://127.0.0.1:9/token".into(),
3507 chrono::Duration::minutes(30),
3508 );
3509 h.repository
3510 .store_token(id, &credential.encode().unwrap())
3511 .await
3512 .unwrap();
3513
3514 let editing = h.provider_service.provider_for_editing(id).await.unwrap();
3515 assert_eq!(editing.token, "");
3516 assert_eq!(
3517 editing.config.get(SIGN_IN_CONFIG_KEY),
3518 Some(&format!("provider:{id}"))
3519 );
3520
3521 h.provider_service
3522 .update_provider(id, editing)
3523 .await
3524 .unwrap();
3525
3526 assert_eq!(stored_sign_in(&h, id).await, credential);
3527 let other = h.add_provider("other").await;
3528 let mut stolen = h
3529 .provider_service
3530 .provider_for_editing(other)
3531 .await
3532 .unwrap();
3533 stolen
3534 .config
3535 .insert(SIGN_IN_CONFIG_KEY.into(), format!("provider:{id}"));
3536 assert!(matches!(
3537 h.provider_service.update_provider(other, stolen).await,
3538 Err(DomainError::InvalidConfig(_))
3539 ));
3540 }
3541
3542 #[tokio::test]
3543 async fn editing_during_a_renewal_keeps_the_renewed_session() {
3544 let h = Harness::with_real_plugins().await;
3545 let argocd = FeedStub::start().await;
3546 argocd.respond_json(serde_json::json!({ "items": [] }));
3547 let original = sign_in(
3548 "http://127.0.0.1:9/token".into(),
3549 chrono::Duration::minutes(30),
3550 );
3551 let mut config = h.config("sso", 30);
3552 config
3553 .config
3554 .insert("server_url".into(), argocd.base_url.clone());
3555 config.token = original.encode().unwrap();
3556 let id = h.repository.add_provider(&config).await.unwrap();
3557 let editing = h.provider_service.provider_for_editing(id).await.unwrap();
3558
3559 let checks = Gate::new();
3560 argocd.set_gate(Some(&checks));
3561 let edit = tokio::spawn({
3562 let service = Arc::clone(&h.provider_service);
3563 async move { service.update_provider(id, editing).await }
3564 });
3565 checks.arrived().await;
3566 let renewed = SignInCredential {
3567 token: "renewed".into(),
3568 refresh_token: Some("r2".into()),
3569 expires_at: Some(Utc::now() + chrono::Duration::minutes(45)),
3570 ..original.clone()
3571 };
3572 h.repository
3573 .store_token(id, &renewed.encode().unwrap())
3574 .await
3575 .unwrap();
3576 checks.open();
3577 edit.await.unwrap().unwrap();
3578
3579 assert_eq!(stored_sign_in(&h, id).await, renewed);
3580 assert_eq!(
3581 h.provider_service
3582 .acquire(id)
3583 .await
3584 .unwrap()
3585 .provider
3586 .renew_at(),
3587 renewed.renew_at()
3588 );
3589 }
3590}