1use std::collections::{
2 BTreeMap,
3 HashMap,
4};
5use std::future::Future;
6use std::sync::{
7 Arc,
8 Mutex,
9 PoisonError,
10};
11use std::time::Duration;
12
13use pipedash_plugin_api::{
14 AvailablePipeline,
15 Organization,
16 PaginatedAvailablePipelines,
17 PaginatedResponse,
18 PaginationParams,
19 PipelineCatalog,
20 Plugin as PluginTrait,
21 PluginError,
22};
23use tokio::sync::OnceCell;
24use tokio::time::Instant;
25
26use super::ProviderService;
27use crate::domain::{
28 DomainError,
29 DomainResult,
30};
31use crate::infrastructure::providers::PluginAdapter;
32
33const CATALOG_TTL: Duration = Duration::from_secs(300);
34const CATALOG_MAX_ENTRIES: usize = 8;
35const DEFAULT_PAGE_SIZE: usize = 100;
36const SELECTION_KEY: &str = "selected_items";
37
38enum Miss {
39 Unsupported(Box<dyn PluginTrait>),
40 Failed(DomainError),
41}
42
43#[derive(Clone, PartialEq, Eq, Hash)]
44struct CatalogKey {
45 provider_type: String,
46 config: BTreeMap<String, String>,
47}
48
49impl CatalogKey {
50 fn new(provider_type: &str, config: &HashMap<String, String>) -> Self {
51 Self {
52 provider_type: provider_type.to_owned(),
53 config: config
54 .iter()
55 .filter(|(key, _)| key.as_str() != SELECTION_KEY)
56 .map(|(key, value)| (key.clone(), value.clone()))
57 .collect(),
58 }
59 }
60}
61
62struct Loaded {
63 catalog: Arc<PipelineCatalog>,
64 at: Instant,
65}
66
67struct Entry {
68 cell: Arc<OnceCell<Loaded>>,
69 seq: u64,
70}
71
72#[derive(Default)]
73struct State {
74 entries: HashMap<CatalogKey, Entry>,
75 next_seq: u64,
76}
77
78pub(super) struct DiscoveryCache {
79 ttl: Duration,
80 max_entries: usize,
81 state: Mutex<State>,
82}
83
84impl Default for DiscoveryCache {
85 fn default() -> Self {
86 Self::new(CATALOG_TTL, CATALOG_MAX_ENTRIES)
87 }
88}
89
90impl DiscoveryCache {
91 fn new(ttl: Duration, max_entries: usize) -> Self {
92 Self {
93 ttl,
94 max_entries,
95 state: Mutex::new(State::default()),
96 }
97 }
98
99 async fn get_or_fetch<E, F, Fut>(
100 &self, key: CatalogKey, fetch: F,
101 ) -> Result<Arc<PipelineCatalog>, E>
102 where
103 F: FnOnce() -> Fut,
104 Fut: Future<Output = Result<PipelineCatalog, E>>,
105 {
106 let cell = self.cell(&key);
107 let loaded = cell
108 .get_or_try_init(|| async {
109 fetch().await.map(|catalog| Loaded {
110 catalog: Arc::new(catalog),
111 at: Instant::now(),
112 })
113 })
114 .await;
115
116 match loaded {
117 Ok(loaded) => Ok(Arc::clone(&loaded.catalog)),
118 Err(e) => {
119 self.discard(&key, &cell);
120 Err(e)
121 }
122 }
123 }
124
125 fn cell(&self, key: &CatalogKey) -> Arc<OnceCell<Loaded>> {
126 let now = Instant::now();
127 let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
128
129 if let Some(entry) = state.entries.get(key) {
130 if !self.expired(entry, now) {
131 return Arc::clone(&entry.cell);
132 }
133 }
134
135 let ttl = self.ttl;
136 state.entries.retain(|_, entry| {
137 entry
138 .cell
139 .get()
140 .is_none_or(|loaded| now.duration_since(loaded.at) < ttl)
141 });
142 while state.entries.len() >= self.max_entries {
143 let Some(oldest) = state
144 .entries
145 .iter()
146 .min_by_key(|(_, entry)| entry.seq)
147 .map(|(key, _)| key.clone())
148 else {
149 break;
150 };
151 state.entries.remove(&oldest);
152 }
153
154 let cell = Arc::new(OnceCell::new());
155 let seq = state.next_seq;
156 state.next_seq += 1;
157 state.entries.insert(
158 key.clone(),
159 Entry {
160 cell: Arc::clone(&cell),
161 seq,
162 },
163 );
164 cell
165 }
166
167 fn expired(&self, entry: &Entry, now: Instant) -> bool {
168 entry
169 .cell
170 .get()
171 .is_some_and(|loaded| now.duration_since(loaded.at) >= self.ttl)
172 }
173
174 fn discard(&self, key: &CatalogKey, cell: &Arc<OnceCell<Loaded>>) {
175 let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
176 if state
177 .entries
178 .get(key)
179 .is_some_and(|entry| Arc::ptr_eq(&entry.cell, cell) && !cell.initialized())
180 {
181 state.entries.remove(key);
182 }
183 }
184}
185
186fn matches_search(pipeline: &AvailablePipeline, needle: &str) -> bool {
187 [
188 Some(pipeline.name.as_str()),
189 pipeline.repository.as_deref(),
190 pipeline.description.as_deref(),
191 ]
192 .into_iter()
193 .flatten()
194 .any(|field| field.to_lowercase().contains(needle))
195}
196
197fn serve_pipelines(
198 catalog: &PipelineCatalog, org: Option<&str>, search: Option<&str>, params: &PaginationParams,
199) -> PaginatedAvailablePipelines {
200 let needle = search
201 .map(str::trim)
202 .filter(|search| !search.is_empty())
203 .map(str::to_lowercase);
204 let mut matched: Vec<&AvailablePipeline> = catalog
205 .pipelines
206 .iter()
207 .filter(|pipeline| org.is_none_or(|org| pipeline.organization.as_deref() == Some(org)))
208 .filter(|pipeline| {
209 needle
210 .as_deref()
211 .is_none_or(|needle| matches_search(pipeline, needle))
212 })
213 .collect();
214 matched.sort_by(|a, b| a.name.cmp(&b.name));
215
216 let total = matched.len();
217 let page = params.page.max(1);
218 let page_size = match params.page_size {
219 0 => DEFAULT_PAGE_SIZE,
220 size => size,
221 };
222 let offset = (page - 1).saturating_mul(page_size);
223 let items: Vec<AvailablePipeline> = matched
224 .into_iter()
225 .skip(offset)
226 .take(page_size)
227 .cloned()
228 .collect();
229 let has_more = offset.saturating_add(items.len()) < total;
230
231 let mut response = PaginatedResponse::new(items, page, page_size, total);
232 response.has_more = has_more;
233 response
234}
235
236impl ProviderService {
237 pub async fn discover_organizations(
238 &self, provider_type: &str, config: HashMap<String, String>,
239 ) -> DomainResult<Vec<Organization>> {
240 match self.catalog(provider_type, config).await {
241 Ok(catalog) => Ok(catalog.organizations.clone()),
242 Err(Miss::Unsupported(plugin)) => plugin
243 .fetch_organizations()
244 .await
245 .map_err(PluginAdapter::map_error),
246 Err(Miss::Failed(e)) => Err(e),
247 }
248 }
249
250 pub async fn discover_pipelines(
251 &self, provider_type: &str, config: HashMap<String, String>, org: Option<String>,
252 search: Option<String>, params: PaginationParams,
253 ) -> DomainResult<PaginatedAvailablePipelines> {
254 match self.catalog(provider_type, config).await {
255 Ok(catalog) => Ok(serve_pipelines(
256 &catalog,
257 org.as_deref(),
258 search.as_deref(),
259 ¶ms,
260 )),
261 Err(Miss::Unsupported(plugin)) => plugin
262 .fetch_available_pipelines_filtered(org, search, Some(params))
263 .await
264 .map_err(PluginAdapter::map_error),
265 Err(Miss::Failed(e)) => Err(e),
266 }
267 }
268
269 async fn catalog(
270 &self, provider_type: &str, config: HashMap<String, String>,
271 ) -> Result<Arc<PipelineCatalog>, Miss> {
272 let key = CatalogKey::new(provider_type, &config);
273
274 self.discovery
275 .get_or_fetch(key, || async {
276 let plugin = self
277 .init_plugin(provider_type, config, 0)
278 .await
279 .map_err(Miss::Failed)?;
280
281 let (validated, catalog) = tokio::join!(
282 plugin.validate_credentials(),
283 plugin.fetch_pipeline_catalog()
284 );
285
286 validated.map_err(|e| Miss::Failed(PluginAdapter::map_error(e)))?;
287
288 match catalog {
289 Ok(catalog) => Ok(catalog),
290 Err(PluginError::NotSupported(_)) => Err(Miss::Unsupported(plugin)),
291 Err(e) => Err(Miss::Failed(PluginAdapter::map_error(e))),
292 }
293 })
294 .await
295 }
296}
297
298#[cfg(test)]
299mod tests {
300 use std::sync::atomic::{
301 AtomicUsize,
302 Ordering,
303 };
304
305 use super::*;
306
307 fn key(provider_type: &str, pairs: &[(&str, &str)]) -> CatalogKey {
308 let config = pairs
309 .iter()
310 .map(|(k, v)| (k.to_string(), v.to_string()))
311 .collect();
312 CatalogKey::new(provider_type, &config)
313 }
314
315 fn pipeline(name: &str, org: &str, repo: &str, description: &str) -> AvailablePipeline {
316 AvailablePipeline {
317 id: format!("{org}/{name}"),
318 name: name.to_string(),
319 description: Some(description.to_string()),
320 organization: Some(org.to_string()),
321 repository: Some(repo.to_string()),
322 }
323 }
324
325 fn catalog_of(names: &[&str]) -> PipelineCatalog {
326 PipelineCatalog {
327 organizations: Vec::new(),
328 pipelines: names
329 .iter()
330 .map(|name| pipeline(name, "org", "repo", ""))
331 .collect(),
332 }
333 }
334
335 async fn counted(
336 calls: &AtomicUsize, result: Result<PipelineCatalog, &'static str>,
337 ) -> Result<PipelineCatalog, &'static str> {
338 calls.fetch_add(1, Ordering::SeqCst);
339 for _ in 0..3 {
340 tokio::task::yield_now().await;
341 }
342 result
343 }
344
345 #[tokio::test]
346 async fn concurrent_misses_on_one_key_fetch_once() {
347 let cache = DiscoveryCache::default();
348 let calls = AtomicUsize::new(0);
349
350 let (a, b) = tokio::join!(
351 cache.get_or_fetch(key("prow", &[("url", "x")]), || counted(
352 &calls,
353 Ok(catalog_of(&["a"]))
354 )),
355 cache.get_or_fetch(key("prow", &[("url", "x")]), || counted(
356 &calls,
357 Ok(catalog_of(&["a"]))
358 )),
359 );
360
361 assert_eq!(calls.load(Ordering::SeqCst), 1);
362 assert!(Arc::ptr_eq(&a.unwrap(), &b.unwrap()));
363 }
364
365 #[tokio::test]
366 async fn hit_does_not_fetch_again() {
367 let cache = DiscoveryCache::default();
368 let calls = AtomicUsize::new(0);
369
370 for _ in 0..3 {
371 cache
372 .get_or_fetch(key("prow", &[]), || counted(&calls, Ok(catalog_of(&[]))))
373 .await
374 .unwrap();
375 }
376
377 assert_eq!(calls.load(Ordering::SeqCst), 1);
378 }
379
380 #[tokio::test]
381 async fn failed_fetch_is_retried_by_the_next_call() {
382 let cache = DiscoveryCache::default();
383 let calls = AtomicUsize::new(0);
384
385 let failed = cache
386 .get_or_fetch(key("prow", &[]), || counted(&calls, Err("boom")))
387 .await;
388 assert_eq!(failed.err(), Some("boom"));
389
390 cache
391 .get_or_fetch(key("prow", &[]), || counted(&calls, Ok(catalog_of(&[]))))
392 .await
393 .unwrap();
394
395 assert_eq!(calls.load(Ordering::SeqCst), 2);
396 assert_eq!(cache.state.lock().unwrap().entries.len(), 1);
397 }
398
399 #[tokio::test(start_paused = true)]
400 async fn expired_entry_is_fetched_again() {
401 let cache = DiscoveryCache::default();
402 let calls = AtomicUsize::new(0);
403 let fetch = || counted(&calls, Ok(catalog_of(&[])));
404
405 cache.get_or_fetch(key("prow", &[]), fetch).await.unwrap();
406 tokio::time::advance(CATALOG_TTL - Duration::from_secs(1)).await;
407 cache.get_or_fetch(key("prow", &[]), fetch).await.unwrap();
408 assert_eq!(calls.load(Ordering::SeqCst), 1);
409
410 tokio::time::advance(Duration::from_secs(1)).await;
411 cache.get_or_fetch(key("prow", &[]), fetch).await.unwrap();
412 assert_eq!(calls.load(Ordering::SeqCst), 2);
413 }
414
415 #[tokio::test]
416 async fn selected_items_does_not_split_the_key() {
417 let cache = DiscoveryCache::default();
418 let calls = AtomicUsize::new(0);
419 let fetch = || counted(&calls, Ok(catalog_of(&[])));
420
421 for selected in ["", "a", "a,b"] {
422 cache
423 .get_or_fetch(
424 key("prow", &[("url", "x"), ("selected_items", selected)]),
425 fetch,
426 )
427 .await
428 .unwrap();
429 }
430 assert_eq!(calls.load(Ordering::SeqCst), 1);
431
432 cache
433 .get_or_fetch(key("prow", &[("url", "y")]), fetch)
434 .await
435 .unwrap();
436 cache
437 .get_or_fetch(key("other", &[("url", "x")]), fetch)
438 .await
439 .unwrap();
440 assert_eq!(calls.load(Ordering::SeqCst), 3);
441 }
442
443 #[tokio::test]
444 async fn oldest_entry_is_dropped_past_the_cap() {
445 let cache = DiscoveryCache::new(CATALOG_TTL, 2);
446 let calls = AtomicUsize::new(0);
447 let fetch = || counted(&calls, Ok(catalog_of(&[])));
448
449 for id in ["1", "2", "3"] {
450 cache
451 .get_or_fetch(key("prow", &[("id", id)]), fetch)
452 .await
453 .unwrap();
454 }
455 assert_eq!(cache.state.lock().unwrap().entries.len(), 2);
456
457 cache
458 .get_or_fetch(key("prow", &[("id", "3")]), fetch)
459 .await
460 .unwrap();
461 assert_eq!(calls.load(Ordering::SeqCst), 3);
462
463 cache
464 .get_or_fetch(key("prow", &[("id", "1")]), fetch)
465 .await
466 .unwrap();
467 assert_eq!(calls.load(Ordering::SeqCst), 4);
468 }
469
470 fn page(page: usize, page_size: usize) -> PaginationParams {
471 PaginationParams { page, page_size }
472 }
473
474 fn names(response: &PaginatedAvailablePipelines) -> Vec<&str> {
475 response.items.iter().map(|p| p.name.as_str()).collect()
476 }
477
478 fn sample() -> PipelineCatalog {
479 PipelineCatalog {
480 organizations: Vec::new(),
481 pipelines: vec![
482 pipeline("zeta", "kube", "kubernetes/kubernetes", "unit tests"),
483 pipeline("alpha", "kube", "kubernetes/test-infra", "Lint"),
484 pipeline("mid", "istio", "istio/istio", "builds the KUBE images"),
485 pipeline("beta", "istio", "istio/api", "e2e"),
486 ],
487 }
488 }
489
490 #[test]
491 fn serving_filters_by_org_and_sorts_by_name() {
492 let response = serve_pipelines(&sample(), Some("kube"), None, &page(1, 10));
493
494 assert_eq!(names(&response), ["alpha", "zeta"]);
495 assert_eq!(response.total_count, 2);
496 assert!(!response.has_more);
497 }
498
499 #[test]
500 fn serving_search_matches_name_repository_and_description() {
501 let by_name = serve_pipelines(&sample(), None, Some(" ALPH "), &page(1, 10));
502 assert_eq!(names(&by_name), ["alpha"]);
503
504 let by_repo = serve_pipelines(&sample(), None, Some("istio/api"), &page(1, 10));
505 assert_eq!(names(&by_repo), ["beta"]);
506
507 let by_description = serve_pipelines(&sample(), None, Some("kube images"), &page(1, 10));
508 assert_eq!(names(&by_description), ["mid"]);
509
510 let org_and_search = serve_pipelines(&sample(), Some("kube"), Some("KUBE"), &page(1, 10));
511 assert_eq!(names(&org_and_search), ["alpha", "zeta"]);
512
513 let none = serve_pipelines(&sample(), Some("istio"), Some("lint"), &page(1, 10));
514 assert!(none.items.is_empty());
515 }
516
517 #[test]
518 fn serving_blank_search_keeps_everything() {
519 let response = serve_pipelines(&sample(), None, Some(" "), &page(1, 10));
520
521 assert_eq!(response.total_count, 4);
522 }
523
524 #[test]
525 fn serving_pages_and_reports_has_more() {
526 let first = serve_pipelines(&sample(), None, None, &page(1, 3));
527 assert_eq!(names(&first), ["alpha", "beta", "mid"]);
528 assert!(first.has_more);
529 assert_eq!(first.total_pages, 2);
530
531 let last = serve_pipelines(&sample(), None, None, &page(2, 3));
532 assert_eq!(names(&last), ["zeta"]);
533 assert!(!last.has_more);
534
535 let past_end = serve_pipelines(&sample(), None, None, &page(3, 3));
536 assert!(past_end.items.is_empty());
537 assert!(!past_end.has_more);
538 }
539
540 #[test]
541 fn serving_exact_last_page_has_no_more() {
542 let first = serve_pipelines(&sample(), None, None, &page(1, 2));
543 assert!(first.has_more);
544
545 let last = serve_pipelines(&sample(), None, None, &page(2, 2));
546 assert_eq!(names(&last), ["mid", "zeta"]);
547 assert_eq!(last.items.len(), last.page_size);
548 assert!(!last.has_more);
549 }
550
551 #[test]
552 fn serving_defaults_page_and_page_size() {
553 let response = serve_pipelines(&sample(), None, None, &page(0, 0));
554
555 assert_eq!(response.page, 1);
556 assert_eq!(response.page_size, DEFAULT_PAGE_SIZE);
557 assert_eq!(response.items.len(), 4);
558 }
559}