1use std::collections::HashMap;
2use std::sync::Arc;
3use std::time::Duration;
4
5use chrono::{
6 DateTime,
7 Utc,
8};
9use sqlx::postgres::PgPool;
10use sqlx::{
11 Row as SqlxRow,
12 SqlitePool,
13};
14use tokio::time::sleep;
15
16use crate::domain::{
17 DomainError,
18 DomainResult,
19 Pipeline,
20 PipelineRun,
21 PipelineStatus,
22 ProviderConfig,
23};
24use crate::infrastructure::deduplication::hash_pipeline_run;
25use crate::infrastructure::{
26 ConfigBackend,
27 ConfigState,
28 TokenStore,
29};
30
31#[derive(Clone)]
32pub enum DatabasePool {
33 Sqlite(SqlitePool),
34 Postgres(PgPool),
35}
36
37impl DatabasePool {
38 pub fn is_postgres(&self) -> bool {
39 matches!(self, DatabasePool::Postgres(_))
40 }
41
42 pub fn as_sqlite(&self) -> Option<&SqlitePool> {
43 match self {
44 DatabasePool::Sqlite(pool) => Some(pool),
45 _ => None,
46 }
47 }
48
49 pub fn as_postgres(&self) -> Option<&PgPool> {
50 match self {
51 DatabasePool::Postgres(pool) => Some(pool),
52 _ => None,
53 }
54 }
55}
56
57pub(crate) const FETCH_STATUS_SUCCESS: &str = "success";
58pub(crate) const FETCH_STATUS_ERROR: &str = "error";
59pub(crate) const FETCH_STATUS_NEVER: &str = "never";
60const WORKFLOW_PARAMETERS_TTL_MINUTES: u32 = 10;
61
62async fn retry_on_busy<F, Fut, T>(operation: F) -> DomainResult<T>
63where
64 F: Fn() -> Fut,
65 Fut: std::future::Future<Output = DomainResult<T>>,
66{
67 const MAX_RETRIES: u32 = 3; const INITIAL_DELAY_MS: u64 = 10;
70
71 let mut attempt = 0;
72 loop {
73 match operation().await {
74 Ok(result) => {
75 if attempt > 0 {
76 tracing::warn!(
77 "SQLite operation succeeded after {} retries ({}ms total)",
78 attempt,
79 INITIAL_DELAY_MS * (2_u64.pow(attempt) - 1)
80 );
81 }
82 return Ok(result);
83 }
84 Err(DomainError::DatabaseError(ref msg))
85 if (msg.contains("database is locked")
86 || msg.contains("SQLITE_BUSY")
87 || msg.contains("locked"))
88 && attempt < MAX_RETRIES =>
89 {
90 attempt += 1;
91 let delay = INITIAL_DELAY_MS * 2_u64.pow(attempt - 1);
92
93 if attempt > 3 {
94 tracing::warn!(
95 "SQLite busy, retry {}/{} (waiting {}ms)",
96 attempt,
97 MAX_RETRIES,
98 delay
99 );
100 }
101
102 sleep(Duration::from_millis(delay)).await;
103 }
104 Err(e) => {
105 if attempt > 0 {
106 tracing::error!("SQLite operation failed after {} retries: {}", attempt, e);
107 }
108 return Err(e);
109 }
110 }
111 }
112}
113
114fn run_order(pipeline_id: &str) -> &'static str {
115 if pipeline_id.starts_with("fluxcd__") {
116 "started_at_ms DESC, run_number DESC"
117 } else {
118 "run_number DESC"
119 }
120}
121
122fn run_cache_row(run: &PipelineRun) -> DomainResult<(i64, String, String, i64)> {
123 let run_data =
124 serde_json::to_string(run).map_err(|e| DomainError::DatabaseError(e.to_string()))?;
125 let run_hash = hash_pipeline_run(
126 run.run_number,
127 run.status.as_str(),
128 run.branch.as_deref(),
129 &run.started_at.to_rfc3339(),
130 run.duration_seconds,
131 run.commit_sha.as_deref(),
132 );
133
134 Ok((
135 run.run_number,
136 run_data,
137 run_hash,
138 run.started_at.timestamp_millis(),
139 ))
140}
141
142pub struct ProviderFetchStatus {
143 pub status: String,
144 pub error: Option<String>,
145 pub fetched_at: Option<DateTime<Utc>>,
146 pub succeeded_at: Option<DateTime<Utc>>,
147}
148
149pub struct Repository {
150 config_backend: Arc<dyn ConfigBackend>,
151 cache_pool: DatabasePool,
152 token_store: Arc<dyn TokenStore>,
153}
154
155impl Repository {
156 pub fn new(
157 config_backend: Arc<dyn ConfigBackend>, cache_pool: DatabasePool,
158 token_store: Arc<dyn TokenStore>,
159 ) -> Self {
160 Self {
161 config_backend,
162 cache_pool,
163 token_store,
164 }
165 }
166
167 pub fn cache_pool(&self) -> &DatabasePool {
168 &self.cache_pool
169 }
170
171 pub fn config_backend(&self) -> &Arc<dyn ConfigBackend> {
172 &self.config_backend
173 }
174
175 pub fn pool(&self) -> &SqlitePool {
176 self.cache_pool
177 .as_sqlite()
178 .expect("Expected SQLite pool but found PostgreSQL")
179 }
180
181 #[allow(dead_code)]
182 pub(crate) fn datetime_now(&self) -> &'static str {
183 match self.cache_pool {
184 DatabasePool::Sqlite(_) => "datetime('now')",
185 DatabasePool::Postgres(_) => "NOW()",
186 }
187 }
188
189 fn placeholder(&self, index: usize) -> String {
190 match self.cache_pool {
191 DatabasePool::Sqlite(_) => "?".to_string(),
192 DatabasePool::Postgres(_) => format!("${}", index),
193 }
194 }
195
196 pub async fn add_provider(&self, config: &ProviderConfig) -> DomainResult<i64> {
197 let provider_id = self.config_backend.create_provider(config).await?;
198
199 self.token_store
200 .store_token(provider_id, &config.token)
201 .await
202 .map_err(|e| {
203 tracing::error!(
204 provider_id = provider_id,
205 error = %e,
206 "Failed to store provider token - provider may not work correctly"
207 );
208 e
209 })?;
210
211 tracing::debug!(
212 provider_id = provider_id,
213 "Provider created and token stored successfully"
214 );
215
216 Ok(provider_id)
217 }
218
219 pub async fn get_provider(&self, id: i64) -> DomainResult<ProviderConfig> {
220 let mut provider = self
221 .config_backend
222 .get_provider(id)
223 .await?
224 .ok_or_else(|| DomainError::ProviderNotFound(id.to_string()))?;
225
226 provider.token = match self.token_store.get_token(id).await {
227 Ok(token) => token,
228 Err(DomainError::NotFound(_)) => String::new(),
229 Err(e) => return Err(e),
230 };
231
232 Ok(provider)
233 }
234
235 pub async fn store_token(&self, id: i64, token: &str) -> DomainResult<()> {
236 self.token_store.store_token(id, token).await
237 }
238
239 pub async fn list_providers(&self) -> DomainResult<Vec<ProviderConfig>> {
240 let mut providers = self.config_backend.list_providers().await?;
241
242 for provider in &mut providers {
243 if let Some(id) = provider.id {
244 provider.token = self.token_store.get_token(id).await.unwrap_or_default();
245 }
246 }
247
248 Ok(providers)
249 }
250
251 pub async fn list_provider_ids(&self) -> DomainResult<Vec<i64>> {
252 Ok(self
253 .config_backend
254 .list_providers()
255 .await?
256 .into_iter()
257 .filter_map(|provider| provider.id)
258 .collect())
259 }
260
261 pub async fn update_provider(&self, id: i64, config: &ProviderConfig) -> DomainResult<()> {
262 self.config_backend.update_provider(id, config).await?;
263
264 self.token_store.store_token(id, &config.token).await?;
265
266 Ok(())
267 }
268
269 pub async fn update_provider_with_version(
270 &self, id: i64, config: &ProviderConfig, expected_version: i64,
271 ) -> DomainResult<bool> {
272 let cache_pool = self.cache_pool.clone();
273 let config_clone = config.clone();
274 let token_clone = config.token.clone();
275 let token_store = self.token_store.clone();
276
277 retry_on_busy(|| async {
278 let config_json = serde_json::to_string(&config_clone.config)
279 .map_err(|e| DomainError::DatabaseError(format!("Failed to serialize config: {}", e)))?;
280
281 let rows_affected = match &cache_pool {
282 DatabasePool::Sqlite(p) => {
283 sqlx::query(
284 r#"UPDATE providers
285 SET name = ?, provider_type = ?, config_json = ?, refresh_interval = ?,
286 version = version + 1, updated_at = datetime('now')
287 WHERE id = ? AND version = ?"#
288 )
289 .bind(&config_clone.name)
290 .bind(&config_clone.provider_type)
291 .bind(&config_json)
292 .bind(config_clone.refresh_interval)
293 .bind(id)
294 .bind(expected_version)
295 .execute(p)
296 .await
297 .map_err(|e| DomainError::DatabaseError(format!("Failed to update provider: {}", e)))?
298 .rows_affected()
299 }
300 DatabasePool::Postgres(p) => {
301 sqlx::query(
302 r#"UPDATE providers
303 SET name = $1, provider_type = $2, config_json = $3, refresh_interval = $4,
304 version = version + 1, updated_at = NOW()
305 WHERE id = $5 AND version = $6"#
306 )
307 .bind(&config_clone.name)
308 .bind(&config_clone.provider_type)
309 .bind(&config_json)
310 .bind(config_clone.refresh_interval)
311 .bind(id)
312 .bind(expected_version)
313 .execute(p)
314 .await
315 .map_err(|e| DomainError::DatabaseError(format!("Failed to update provider: {}", e)))?
316 .rows_affected()
317 }
318 };
319
320 let success = rows_affected > 0;
321
322 if success {
323 token_store.store_token(id, &token_clone).await?;
324 }
325
326 Ok(success)
327 }).await
328 }
329
330 pub async fn remove_provider(&self, id: i64, config_state: &ConfigState) -> DomainResult<()> {
331 let config = self.config_backend.get_provider(id).await?;
332 self.token_store.delete_token(id).await?;
333 if let Some(config) = config {
334 config_state.remove_provider(&config.name).await?;
335 }
336
337 let pipelines = self
338 .get_cached_pipelines(Some(id))
339 .await
340 .unwrap_or_default();
341 let pipeline_ids: Vec<String> = pipelines.iter().map(|p| p.id.clone()).collect();
342
343 let run_history_sql = format!(
344 "DELETE FROM run_history_cache WHERE pipeline_id IN (SELECT id FROM pipelines_cache WHERE provider_id = {})",
345 self.placeholder(1)
346 );
347 match &self.cache_pool {
348 DatabasePool::Sqlite(p) => {
349 let _ = sqlx::query(sqlx::AssertSqlSafe(run_history_sql))
350 .bind(id)
351 .execute(p)
352 .await
353 .map_err(|e| DomainError::DatabaseError(e.to_string()));
354 }
355 DatabasePool::Postgres(p) => {
356 let _ = sqlx::query(sqlx::AssertSqlSafe(run_history_sql))
357 .bind(id)
358 .execute(p)
359 .await
360 .map_err(|e| DomainError::DatabaseError(e.to_string()));
361 }
362 }
363
364 for pipeline_id in pipeline_ids {
365 let workflow_params_sql = format!(
366 "DELETE FROM workflow_parameters_cache WHERE workflow_id LIKE {}",
367 self.placeholder(1)
368 );
369 match &self.cache_pool {
370 DatabasePool::Sqlite(p) => {
371 let _ = sqlx::query(sqlx::AssertSqlSafe(workflow_params_sql))
372 .bind(format!("{}%", pipeline_id))
373 .execute(p)
374 .await
375 .map_err(|e| DomainError::DatabaseError(e.to_string()));
376 }
377 DatabasePool::Postgres(p) => {
378 let _ = sqlx::query(sqlx::AssertSqlSafe(workflow_params_sql))
379 .bind(format!("{}%", pipeline_id))
380 .execute(p)
381 .await
382 .map_err(|e| DomainError::DatabaseError(e.to_string()));
383 }
384 }
385 }
386
387 let pipelines_sql = format!(
388 "DELETE FROM pipelines_cache WHERE provider_id = {}",
389 self.placeholder(1)
390 );
391 match &self.cache_pool {
392 DatabasePool::Sqlite(p) => {
393 let _ = sqlx::query(sqlx::AssertSqlSafe(pipelines_sql))
394 .bind(id)
395 .execute(p)
396 .await
397 .map_err(|e| DomainError::DatabaseError(e.to_string()));
398 }
399 DatabasePool::Postgres(p) => {
400 let _ = sqlx::query(sqlx::AssertSqlSafe(pipelines_sql))
401 .bind(id)
402 .execute(p)
403 .await
404 .map_err(|e| DomainError::DatabaseError(e.to_string()));
405 }
406 }
407
408 if let Err(e) = super::PipelineViewRepository::new(self.cache_pool.clone())
409 .remove_provider(id)
410 .await
411 {
412 tracing::warn!("Failed to remove provider {id} from pipeline views: {e}");
413 }
414
415 self.config_backend.delete_provider(id).await?;
416
417 Ok(())
418 }
419
420 pub async fn update_provider_fetch_status(
421 &self, provider_id: i64, success: bool, error: Option<String>,
422 ) -> DomainResult<bool> {
423 let cache_pool = self.cache_pool.clone();
424 let error_clone = error.clone();
425
426 retry_on_busy(|| async {
427 let now = chrono::Utc::now();
428 let new_status = if success {
429 FETCH_STATUS_SUCCESS
430 } else {
431 FETCH_STATUS_ERROR
432 };
433
434 let select_sql = match cache_pool {
435 DatabasePool::Sqlite(_) => {
436 "SELECT last_fetch_status, last_fetch_error FROM providers WHERE id = ?"
437 }
438 DatabasePool::Postgres(_) => {
439 "SELECT last_fetch_status, last_fetch_error FROM providers WHERE id = $1"
440 }
441 };
442
443 let status_changed = match &cache_pool {
444 DatabasePool::Sqlite(p) => {
445 let current = sqlx::query(select_sql)
446 .bind(provider_id)
447 .fetch_optional(p)
448 .await
449 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
450
451 if let Some(row) = current {
452 let current_status: String =
453 row.try_get("last_fetch_status").unwrap_or_default();
454 let current_error: Option<String> =
455 row.try_get("last_fetch_error").ok().flatten();
456 current_status != new_status || current_error != error_clone
457 } else {
458 true
459 }
460 }
461 DatabasePool::Postgres(p) => {
462 let current = sqlx::query(select_sql)
463 .bind(provider_id)
464 .fetch_optional(p)
465 .await
466 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
467
468 if let Some(row) = current {
469 let current_status: String =
470 row.try_get("last_fetch_status").unwrap_or_default();
471 let current_error: Option<String> =
472 row.try_get("last_fetch_error").ok().flatten();
473 current_status != new_status || current_error != error_clone
474 } else {
475 true
476 }
477 }
478 };
479
480 match &cache_pool {
481 DatabasePool::Sqlite(p) => {
482 let update_sql = "UPDATE providers SET last_fetch_at = ?, last_fetch_status = ?, last_fetch_error = ?, last_success_at = COALESCE(?, last_success_at) WHERE id = ?";
483 sqlx::query(update_sql)
484 .bind(now.to_rfc3339())
485 .bind(new_status)
486 .bind(error_clone.as_ref())
487 .bind(success.then(|| now.to_rfc3339()))
488 .bind(provider_id)
489 .execute(p)
490 .await
491 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
492 }
493 DatabasePool::Postgres(p) => {
494 let update_sql = "UPDATE providers SET last_fetch_at = $1, last_fetch_status = $2, last_fetch_error = $3, last_success_at = COALESCE($4, last_success_at) WHERE id = $5";
495 sqlx::query(update_sql)
496 .bind(now)
497 .bind(new_status)
498 .bind(error_clone.as_ref())
499 .bind(success.then_some(now))
500 .bind(provider_id)
501 .execute(p)
502 .await
503 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
504 }
505 }
506
507 Ok(status_changed)
508 }).await
509 }
510
511 pub async fn get_provider_fetch_status(
512 &self, provider_id: i64,
513 ) -> DomainResult<ProviderFetchStatus> {
514 let sql = format!(
515 "SELECT last_fetch_status, last_fetch_error, last_fetch_at, last_success_at FROM providers WHERE id = {}",
516 self.placeholder(1)
517 );
518 let not_found = |_| DomainError::ProviderNotFound(provider_id.to_string());
519 let (row_status, error, fetched_at, succeeded_at) = match &self.cache_pool {
520 DatabasePool::Sqlite(p) => {
521 let row = sqlx::query(sqlx::AssertSqlSafe(sql))
522 .bind(provider_id)
523 .fetch_one(p)
524 .await
525 .map_err(not_found)?;
526 let time = |index: usize| {
527 row.try_get::<Option<String>, _>(index)
528 .ok()
529 .flatten()
530 .and_then(|at| DateTime::parse_from_rfc3339(&at).ok())
531 .map(|at| at.with_timezone(&Utc))
532 };
533
534 (
535 row.try_get::<String, _>(0).ok(),
536 row.try_get(1).ok().flatten(),
537 time(2),
538 time(3),
539 )
540 }
541 DatabasePool::Postgres(p) => {
542 let row = sqlx::query(sqlx::AssertSqlSafe(sql))
543 .bind(provider_id)
544 .fetch_one(p)
545 .await
546 .map_err(not_found)?;
547
548 (
549 row.try_get::<String, _>(0).ok(),
550 row.try_get(1).ok().flatten(),
551 row.try_get(2).ok().flatten(),
552 row.try_get(3).ok().flatten(),
553 )
554 }
555 };
556
557 Ok(ProviderFetchStatus {
558 status: row_status.unwrap_or_else(|| FETCH_STATUS_NEVER.to_string()),
559 error,
560 fetched_at,
561 succeeded_at,
562 })
563 }
564
565 pub async fn reset_provider_fetch_status(&self, provider_id: i64) -> DomainResult<()> {
566 let cache_pool = self.cache_pool.clone();
567
568 retry_on_busy(|| async {
569 match &cache_pool {
570 DatabasePool::Sqlite(p) => {
571 sqlx::query(
572 "UPDATE providers SET last_fetch_at = NULL, last_success_at = NULL, last_fetch_status = ?, last_fetch_error = NULL WHERE id = ?",
573 )
574 .bind(FETCH_STATUS_NEVER)
575 .bind(provider_id)
576 .execute(p)
577 .await
578 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
579 }
580 DatabasePool::Postgres(p) => {
581 sqlx::query(
582 "UPDATE providers SET last_fetch_at = NULL, last_success_at = NULL, last_fetch_status = $1, last_fetch_error = NULL WHERE id = $2",
583 )
584 .bind(FETCH_STATUS_NEVER)
585 .bind(provider_id)
586 .execute(p)
587 .await
588 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
589 }
590 }
591
592 Ok(())
593 })
594 .await
595 }
596
597 pub async fn clear_provider_run_history(&self, provider_id: i64) -> DomainResult<()> {
598 let sql = format!(
599 "DELETE FROM run_history_cache WHERE pipeline_id IN (SELECT id FROM pipelines_cache WHERE provider_id = {})",
600 self.placeholder(1)
601 );
602 match &self.cache_pool {
603 DatabasePool::Sqlite(p) => {
604 sqlx::query(sqlx::AssertSqlSafe(sql))
605 .bind(provider_id)
606 .execute(p)
607 .await
608 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
609 }
610 DatabasePool::Postgres(p) => {
611 sqlx::query(sqlx::AssertSqlSafe(sql))
612 .bind(provider_id)
613 .execute(p)
614 .await
615 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
616 }
617 }
618
619 Ok(())
620 }
621
622 pub async fn get_cached_pipelines(
623 &self, provider_id: Option<i64>,
624 ) -> DomainResult<Vec<Pipeline>> {
625 if let Some(pid) = provider_id {
626 let sql = format!(
627 "SELECT id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, provider_type, metadata_json, resource_state_json
628 FROM pipelines_cache
629 WHERE provider_id = {}
630 ORDER BY last_updated DESC",
631 self.placeholder(1)
632 );
633 match &self.cache_pool {
634 DatabasePool::Sqlite(p) => {
635 let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
636 .bind(pid)
637 .fetch_all(p)
638 .await
639 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
640 rows.iter().map(|row| self.pipeline_from_row(row)).collect()
641 }
642 DatabasePool::Postgres(p) => {
643 let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
644 .bind(pid)
645 .fetch_all(p)
646 .await
647 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
648 rows.iter().map(|row| self.pipeline_from_row(row)).collect()
649 }
650 }
651 } else {
652 let sql = "SELECT id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, provider_type, metadata_json, resource_state_json
653 FROM pipelines_cache
654 ORDER BY last_updated DESC";
655 match &self.cache_pool {
656 DatabasePool::Sqlite(p) => {
657 let rows = sqlx::query(sql)
658 .fetch_all(p)
659 .await
660 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
661 rows.iter().map(|row| self.pipeline_from_row(row)).collect()
662 }
663 DatabasePool::Postgres(p) => {
664 let rows = sqlx::query(sql)
665 .fetch_all(p)
666 .await
667 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
668 rows.iter().map(|row| self.pipeline_from_row(row)).collect()
669 }
670 }
671 }
672 }
673
674 pub async fn cache_workflow_parameters(
675 &self, workflow_id: &str, parameters: &[pipedash_plugin_api::WorkflowParameter],
676 ) -> DomainResult<()> {
677 let workflow_id_owned = workflow_id.to_string();
678 let parameters_vec = parameters.to_vec();
679 let cache_pool = self.cache_pool.clone();
680 let is_postgres = cache_pool.is_postgres();
681
682 retry_on_busy(|| async {
683 let parameters_json = serde_json::to_string(¶meters_vec)
684 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
685
686 let sql = if is_postgres {
687 "INSERT INTO workflow_parameters_cache (workflow_id, parameters_json, cached_at) VALUES ($1, $2, NOW()) ON CONFLICT (workflow_id) DO UPDATE SET parameters_json = EXCLUDED.parameters_json, cached_at = EXCLUDED.cached_at"
688 } else {
689 "INSERT OR REPLACE INTO workflow_parameters_cache (workflow_id, parameters_json, cached_at) VALUES (?, ?, datetime('now'))"
690 };
691
692 match &cache_pool {
693 DatabasePool::Sqlite(p) => {
694 sqlx::query(sql)
695 .bind(&workflow_id_owned)
696 .bind(¶meters_json)
697 .execute(p)
698 .await
699 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
700 }
701 DatabasePool::Postgres(p) => {
702 sqlx::query(sql)
703 .bind(&workflow_id_owned)
704 .bind(¶meters_json)
705 .execute(p)
706 .await
707 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
708 }
709 }
710
711 Ok(())
712 }).await
713 }
714
715 pub async fn get_cached_workflow_parameters(
716 &self, workflow_id: &str,
717 ) -> DomainResult<Option<Vec<pipedash_plugin_api::WorkflowParameter>>> {
718 let freshness = match self.cache_pool {
719 DatabasePool::Sqlite(_) => format!(
720 "cached_at >= datetime('now', '-{WORKFLOW_PARAMETERS_TTL_MINUTES} minutes')"
721 ),
722 DatabasePool::Postgres(_) => {
723 format!("cached_at >= NOW() - INTERVAL '{WORKFLOW_PARAMETERS_TTL_MINUTES} minutes'")
724 }
725 };
726 let sql = format!(
727 "SELECT parameters_json FROM workflow_parameters_cache WHERE workflow_id = {} AND {freshness}",
728 self.placeholder(1)
729 );
730 let result = match &self.cache_pool {
731 DatabasePool::Sqlite(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
732 .bind(workflow_id)
733 .fetch_optional(p)
734 .await
735 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
736 DatabasePool::Postgres(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
737 .bind(workflow_id)
738 .fetch_optional(p)
739 .await
740 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
741 };
742
743 match result {
744 Some(json) => {
745 if json.trim().is_empty() {
746 Ok(Some(vec![]))
747 } else {
748 let parameters: Vec<pipedash_plugin_api::WorkflowParameter> =
749 serde_json::from_str(&json).map_err(|e| {
750 DomainError::DatabaseError(format!(
751 "Failed to parse workflow parameters: {}",
752 e
753 ))
754 })?;
755 Ok(Some(parameters))
756 }
757 }
758 None => Ok(None),
759 }
760 }
761
762 pub async fn clear_workflow_parameters_cache(&self) -> DomainResult<()> {
763 match &self.cache_pool {
764 DatabasePool::Sqlite(p) => {
765 sqlx::query("DELETE FROM workflow_parameters_cache")
766 .execute(p)
767 .await
768 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
769 }
770 DatabasePool::Postgres(p) => {
771 sqlx::query("DELETE FROM workflow_parameters_cache")
772 .execute(p)
773 .await
774 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
775 }
776 }
777
778 Ok(())
779 }
780
781 pub async fn cache_run_history(
782 &self, pipeline_id: &str, runs: &[PipelineRun],
783 ) -> DomainResult<()> {
784 if runs.is_empty() {
785 return Ok(());
786 }
787
788 let start = std::time::Instant::now();
789 let pipeline_id_str = pipeline_id.to_string();
790 let runs_vec = runs.to_vec();
791 let pool = self.cache_pool.clone();
792
793 retry_on_busy(|| {
794 let pipeline_id_clone = pipeline_id_str.clone();
795 let runs_clone = runs_vec.clone();
796 let pool_clone = pool.clone();
797 async move {
798
799 const BATCH_SIZE: usize = 100;
800
801 match &pool_clone {
802 DatabasePool::Sqlite(p) => {
803 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
804
805 for chunk in runs_clone.chunks(BATCH_SIZE) {
806 let prepared_data = chunk
807 .iter()
808 .map(run_cache_row)
809 .collect::<DomainResult<Vec<_>>>()?;
810
811 let values_clause = prepared_data
812 .iter()
813 .map(|_| "(?, ?, ?, datetime('now'), ?, ?)")
814 .collect::<Vec<_>>()
815 .join(", ");
816
817 let sql = format!(
818 "INSERT OR REPLACE INTO run_history_cache (pipeline_id, run_number, run_data, fetched_at, run_hash, started_at_ms) VALUES {}",
819 values_clause
820 );
821
822 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
823 for (run_number, run_data, run_hash, started_at_ms) in &prepared_data {
824 query = query
825 .bind(&pipeline_id_clone)
826 .bind(run_number)
827 .bind(run_data)
828 .bind(run_hash)
829 .bind(started_at_ms);
830 }
831
832 query.execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
833 }
834
835 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
836 }
837 DatabasePool::Postgres(p) => {
838 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
839
840 for chunk in runs_clone.chunks(BATCH_SIZE) {
841 let prepared_data = chunk
842 .iter()
843 .map(run_cache_row)
844 .collect::<DomainResult<Vec<_>>>()?;
845
846 let mut param_idx = 0;
847 let values_clauses: Vec<String> = prepared_data
848 .iter()
849 .map(|_| {
850 let clause = format!(
851 "(${}, ${}, ${}, NOW(), ${}, ${})",
852 param_idx + 1, param_idx + 2, param_idx + 3, param_idx + 4, param_idx + 5
853 );
854 param_idx += 5;
855 clause
856 })
857 .collect();
858
859 let sql = format!(
860 "INSERT INTO run_history_cache (pipeline_id, run_number, run_data, fetched_at, run_hash, started_at_ms) VALUES {} ON CONFLICT (pipeline_id, run_number) DO UPDATE SET run_data = EXCLUDED.run_data, fetched_at = EXCLUDED.fetched_at, run_hash = EXCLUDED.run_hash, started_at_ms = EXCLUDED.started_at_ms",
861 values_clauses.join(", ")
862 );
863
864 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
865 for (run_number, run_data, run_hash, started_at_ms) in &prepared_data {
866 query = query
867 .bind(&pipeline_id_clone)
868 .bind(run_number)
869 .bind(run_data)
870 .bind(run_hash)
871 .bind(started_at_ms);
872 }
873
874 query.execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
875 }
876
877 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
878 }
879 }
880
881 Ok(())
882 }
883 })
884 .await?;
885
886 let elapsed = start.elapsed();
887 tracing::debug!(
888 pipeline_id = pipeline_id,
889 runs_count = runs.len(),
890 elapsed_ms = elapsed.as_millis(),
891 "Cached run history (batch insert)"
892 );
893
894 Ok(())
895 }
896
897 pub(crate) async fn list_run_history_pipeline_ids(&self) -> DomainResult<Vec<String>> {
898 let sql = "SELECT DISTINCT pipeline_id FROM run_history_cache";
899 match &self.cache_pool {
900 DatabasePool::Sqlite(pool) => sqlx::query_scalar(sql).fetch_all(pool).await,
901 DatabasePool::Postgres(pool) => sqlx::query_scalar(sql).fetch_all(pool).await,
902 }
903 .map_err(|error| DomainError::DatabaseError(error.to_string()))
904 }
905
906 pub async fn get_cached_run_history(
907 &self, pipeline_id: &str, limit: usize,
908 ) -> DomainResult<Vec<PipelineRun>> {
909 let sql = format!(
910 "SELECT run_data FROM run_history_cache
911 WHERE pipeline_id = {}
912 ORDER BY {}
913 LIMIT {}",
914 self.placeholder(1),
915 run_order(pipeline_id),
916 self.placeholder(2)
917 );
918 let rows = match &self.cache_pool {
919 DatabasePool::Sqlite(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
920 .bind(pipeline_id)
921 .bind(limit as i64)
922 .fetch_all(p)
923 .await
924 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
925 DatabasePool::Postgres(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
926 .bind(pipeline_id)
927 .bind(limit as i64)
928 .fetch_all(p)
929 .await
930 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
931 };
932
933 let runs: Vec<PipelineRun> = rows
934 .iter()
935 .filter_map(|json| serde_json::from_str(json).ok())
936 .collect();
937
938 Ok(runs)
939 }
940
941 pub async fn get_cached_run(
942 &self, pipeline_id: &str, run_number: i64,
943 ) -> DomainResult<Option<PipelineRun>> {
944 let sql = format!(
945 "SELECT run_data FROM run_history_cache WHERE pipeline_id = {} AND run_number = {}",
946 self.placeholder(1),
947 self.placeholder(2)
948 );
949 let row = match &self.cache_pool {
950 DatabasePool::Sqlite(pool) => {
951 sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
952 .bind(pipeline_id)
953 .bind(run_number)
954 .fetch_optional(pool)
955 .await
956 }
957 DatabasePool::Postgres(pool) => {
958 sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
959 .bind(pipeline_id)
960 .bind(run_number)
961 .fetch_optional(pool)
962 .await
963 }
964 }
965 .map_err(|error| DomainError::DatabaseError(error.to_string()))?;
966
967 row.map(|json| serde_json::from_str(&json))
968 .transpose()
969 .map_err(|error| DomainError::DatabaseError(format!("Invalid cached run: {error}")))
970 }
971
972 pub async fn clear_cached_run_history(&self, pipeline_id: &str) -> DomainResult<()> {
973 let sql = format!(
974 "DELETE FROM run_history_cache WHERE pipeline_id = {}",
975 self.placeholder(1)
976 );
977 match &self.cache_pool {
978 DatabasePool::Sqlite(p) => {
979 sqlx::query(sqlx::AssertSqlSafe(sql))
980 .bind(pipeline_id)
981 .execute(p)
982 .await
983 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
984 }
985 DatabasePool::Postgres(p) => {
986 sqlx::query(sqlx::AssertSqlSafe(sql))
987 .bind(pipeline_id)
988 .execute(p)
989 .await
990 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
991 }
992 }
993
994 Ok(())
995 }
996
997 pub async fn clear_all_run_history_cache(&self) -> DomainResult<()> {
998 match &self.cache_pool {
999 DatabasePool::Sqlite(p) => {
1000 sqlx::query("DELETE FROM run_history_cache")
1001 .execute(p)
1002 .await
1003 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1004 }
1005 DatabasePool::Postgres(p) => {
1006 sqlx::query("DELETE FROM run_history_cache")
1007 .execute(p)
1008 .await
1009 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1010 }
1011 }
1012
1013 Ok(())
1014 }
1015
1016 pub async fn get_cached_runs_with_hashes(
1017 &self, pipeline_id: &str,
1018 ) -> DomainResult<HashMap<i64, (PipelineRun, String)>> {
1019 let mut result = HashMap::new();
1020 match &self.cache_pool {
1021 DatabasePool::Sqlite(p) => {
1022 let sql = format!(
1023 "SELECT run_number, run_data, run_hash FROM run_history_cache WHERE pipeline_id = {}",
1024 self.placeholder(1)
1025 );
1026 let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
1027 .bind(pipeline_id)
1028 .fetch_all(p)
1029 .await
1030 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1031
1032 for row in rows {
1033 let run_number: i64 = row
1034 .try_get(0)
1035 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1036 let run_data: String = row
1037 .try_get(1)
1038 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1039 let run_hash: String = row
1040 .try_get(2)
1041 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1042
1043 if let Ok(run) = serde_json::from_str::<PipelineRun>(&run_data) {
1044 result.insert(run_number, (run, run_hash));
1045 }
1046 }
1047 }
1048 DatabasePool::Postgres(p) => {
1049 let sql = format!(
1050 "SELECT run_number::BIGINT, run_data, run_hash FROM run_history_cache WHERE pipeline_id = {}",
1051 self.placeholder(1)
1052 );
1053 let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
1054 .bind(pipeline_id)
1055 .fetch_all(p)
1056 .await
1057 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1058
1059 for row in rows {
1060 let run_number: i64 = row
1061 .try_get(0)
1062 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1063 let run_data: String = row
1064 .try_get(1)
1065 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1066 let run_hash: String = row
1067 .try_get(2)
1068 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1069
1070 if let Ok(run) = serde_json::from_str::<PipelineRun>(&run_data) {
1071 result.insert(run_number, (run, run_hash));
1072 }
1073 }
1074 }
1075 }
1076
1077 Ok(result)
1078 }
1079
1080 pub async fn merge_run_cache(
1081 &self, pipeline_id: &str, new_runs: Vec<PipelineRun>, changed_runs: Vec<PipelineRun>,
1082 deleted_run_numbers: Vec<i64>,
1083 ) -> DomainResult<()> {
1084 let start = std::time::Instant::now();
1085 let pipeline_id_str = pipeline_id.to_string();
1086 let pool = self.cache_pool.clone();
1087
1088 let new_runs_count = new_runs.len();
1089 let changed_runs_count = changed_runs.len();
1090 let deleted_runs_count = deleted_run_numbers.len();
1091
1092 retry_on_busy(move || {
1093 let pipeline_id_clone = pipeline_id_str.clone();
1094 let new_runs_clone = new_runs.clone();
1095 let changed_runs_clone = changed_runs.clone();
1096 let deleted_clone = deleted_run_numbers.clone();
1097 let pool_clone = pool.clone();
1098
1099 async move {
1100 const BATCH_SIZE: usize = 100;
1101 const DELETE_BATCH_SIZE: usize = 100;
1102
1103 let all_runs: Vec<_> = new_runs_clone.iter().chain(changed_runs_clone.iter()).collect();
1104
1105 match &pool_clone {
1106 DatabasePool::Sqlite(p) => {
1107 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1108
1109 if !all_runs.is_empty() {
1110 for chunk in all_runs.chunks(BATCH_SIZE) {
1111 let prepared_data = chunk
1112 .iter()
1113 .copied()
1114 .map(run_cache_row)
1115 .collect::<DomainResult<Vec<_>>>()?;
1116
1117 let values_clause = prepared_data
1118 .iter()
1119 .map(|_| "(?, ?, ?, datetime('now'), ?, ?)")
1120 .collect::<Vec<_>>()
1121 .join(", ");
1122
1123 let sql = format!(
1124 "INSERT OR REPLACE INTO run_history_cache (pipeline_id, run_number, run_data, fetched_at, run_hash, started_at_ms) VALUES {}",
1125 values_clause
1126 );
1127
1128 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
1129 for (run_number, run_data, run_hash, started_at_ms) in &prepared_data {
1130 query = query
1131 .bind(&pipeline_id_clone)
1132 .bind(run_number)
1133 .bind(run_data)
1134 .bind(run_hash)
1135 .bind(started_at_ms);
1136 }
1137
1138 query.execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1139 }
1140 }
1141
1142 if !deleted_clone.is_empty() {
1143 for chunk in deleted_clone.chunks(DELETE_BATCH_SIZE) {
1144 let placeholders = chunk.iter().map(|_| "?").collect::<Vec<_>>().join(", ");
1145 let sql = format!(
1146 "DELETE FROM run_history_cache WHERE pipeline_id = ? AND run_number IN ({})",
1147 placeholders
1148 );
1149
1150 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(&pipeline_id_clone);
1151 for run_number in chunk {
1152 query = query.bind(run_number);
1153 }
1154
1155 query.execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1156 }
1157 }
1158
1159 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1160 }
1161 DatabasePool::Postgres(p) => {
1162 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1163
1164 if !all_runs.is_empty() {
1165 for chunk in all_runs.chunks(BATCH_SIZE) {
1166 let prepared_data = chunk
1167 .iter()
1168 .copied()
1169 .map(run_cache_row)
1170 .collect::<DomainResult<Vec<_>>>()?;
1171
1172 let mut param_idx = 0;
1173 let values_clauses: Vec<String> = prepared_data
1174 .iter()
1175 .map(|_| {
1176 let clause = format!(
1177 "(${}, ${}, ${}, NOW(), ${}, ${})",
1178 param_idx + 1, param_idx + 2, param_idx + 3, param_idx + 4, param_idx + 5
1179 );
1180 param_idx += 5;
1181 clause
1182 })
1183 .collect();
1184
1185 let sql = format!(
1186 "INSERT INTO run_history_cache (pipeline_id, run_number, run_data, fetched_at, run_hash, started_at_ms) VALUES {} ON CONFLICT (pipeline_id, run_number) DO UPDATE SET run_data = EXCLUDED.run_data, fetched_at = EXCLUDED.fetched_at, run_hash = EXCLUDED.run_hash, started_at_ms = EXCLUDED.started_at_ms",
1187 values_clauses.join(", ")
1188 );
1189
1190 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
1191 for (run_number, run_data, run_hash, started_at_ms) in &prepared_data {
1192 query = query
1193 .bind(&pipeline_id_clone)
1194 .bind(run_number)
1195 .bind(run_data)
1196 .bind(run_hash)
1197 .bind(started_at_ms);
1198 }
1199
1200 query.execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1201 }
1202 }
1203
1204 if !deleted_clone.is_empty() {
1205 for chunk in deleted_clone.chunks(DELETE_BATCH_SIZE) {
1206 let mut param_idx = 1;
1207 let placeholders = chunk.iter().map(|_| {
1208 let p = format!("${}", param_idx);
1209 param_idx += 1;
1210 p
1211 }).collect::<Vec<_>>().join(", ");
1212
1213 let sql = format!(
1214 "DELETE FROM run_history_cache WHERE pipeline_id = ${} AND run_number IN ({})",
1215 param_idx,
1216 placeholders
1217 );
1218
1219 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
1220 for run_number in chunk {
1221 query = query.bind(run_number);
1222 }
1223 query = query.bind(&pipeline_id_clone);
1224
1225 query.execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1226 }
1227 }
1228
1229 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1230 }
1231 }
1232
1233 Ok(())
1234 }
1235 })
1236 .await?;
1237
1238 let elapsed = start.elapsed();
1239 tracing::debug!(
1240 pipeline_id = pipeline_id,
1241 new_runs = new_runs_count,
1242 changed_runs = changed_runs_count,
1243 deleted_runs = deleted_runs_count,
1244 elapsed_ms = elapsed.as_millis(),
1245 "Merged run cache (batch operations)"
1246 );
1247
1248 Ok(())
1249 }
1250
1251 pub async fn update_pipelines_cache(
1252 &self, provider_id: i64, new_pipelines: &[Pipeline],
1253 ) -> DomainResult<()> {
1254 let new_pipelines_vec = new_pipelines.to_vec();
1255 let pool = self.cache_pool.clone();
1256 let is_postgres = pool.is_postgres();
1257
1258 retry_on_busy(move || {
1259 let new_pipelines_clone = new_pipelines_vec.clone();
1260 let pool_clone = pool.clone();
1261 let is_pg = is_postgres;
1262 async move {
1263 let select_sql = if is_pg {
1264 "SELECT id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, provider_type, metadata_json, resource_state_json
1265 FROM pipelines_cache
1266 WHERE provider_id = $1"
1267 } else {
1268 "SELECT id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, provider_type, metadata_json, resource_state_json
1269 FROM pipelines_cache
1270 WHERE provider_id = ?"
1271 };
1272
1273 let update_sql = if is_pg {
1274 "UPDATE pipelines_cache
1275 SET provider_id = $1, name = $2, status = $3, repository = $4, branch = $5,
1276 workflow_file = $6, last_run = $7, last_updated = $8, metadata_json = $9, provider_type = $10,
1277 resource_state_json = $11
1278 WHERE id = $12"
1279 } else {
1280 "UPDATE pipelines_cache
1281 SET provider_id = ?, name = ?, status = ?, repository = ?, branch = ?,
1282 workflow_file = ?, last_run = ?, last_updated = ?, metadata_json = ?, provider_type = ?,
1283 resource_state_json = ?
1284 WHERE id = ?"
1285 };
1286
1287 let insert_sql = if is_pg {
1288 "INSERT INTO pipelines_cache
1289 (id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, metadata_json, provider_type, resource_state_json)
1290 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)"
1291 } else {
1292 "INSERT INTO pipelines_cache
1293 (id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, metadata_json, provider_type, resource_state_json)
1294 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
1295 };
1296
1297 let delete_sql = if is_pg {
1298 "DELETE FROM pipelines_cache WHERE id = $1"
1299 } else {
1300 "DELETE FROM pipelines_cache WHERE id = ?"
1301 };
1302
1303 match &pool_clone {
1304 DatabasePool::Sqlite(p) => {
1305 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1306
1307 let existing_rows = sqlx::query(select_sql)
1308 .bind(provider_id)
1309 .fetch_all(&mut *tx)
1310 .await
1311 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1312
1313 let existing: HashMap<String, Pipeline> = existing_rows.iter()
1314 .map(|row| {
1315 let pipeline = self.pipeline_from_row(row)?;
1316 Ok((pipeline.id.clone(), pipeline))
1317 })
1318 .collect::<DomainResult<_>>()?;
1319
1320 let new_ids: HashMap<String, &Pipeline> =
1321 new_pipelines_clone.iter().map(|p| (p.id.clone(), p)).collect();
1322
1323 for pipeline in &new_pipelines_clone {
1324 if let Some(old) = existing.get(&pipeline.id) {
1325 if old.status != pipeline.status
1326 || old.last_run != pipeline.last_run
1327 || old.name != pipeline.name
1328 || old.provider_id != provider_id
1329 || old.provider_type != pipeline.provider_type
1330 || old.repository != pipeline.repository
1331 || old.branch != pipeline.branch
1332 || old.workflow_file != pipeline.workflow_file
1333 || old.metadata != pipeline.metadata
1334 || old.resource_state != pipeline.resource_state
1335 {
1336 sqlx::query(update_sql)
1337 .bind(provider_id)
1338 .bind(&pipeline.name)
1339 .bind(pipeline.status.as_str())
1340 .bind(&pipeline.repository)
1341 .bind(&pipeline.branch)
1342 .bind(&pipeline.workflow_file)
1343 .bind(pipeline.last_run.as_ref().map(|dt| dt.to_rfc3339()))
1344 .bind(Utc::now().to_rfc3339())
1345 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1346 .bind(&pipeline.provider_type)
1347 .bind(resource_state_json(pipeline)?)
1348 .bind(&pipeline.id)
1349 .execute(&mut *tx)
1350 .await
1351 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1352 }
1353 } else {
1354 sqlx::query(insert_sql)
1355 .bind(&pipeline.id)
1356 .bind(provider_id)
1357 .bind(&pipeline.name)
1358 .bind(pipeline.status.as_str())
1359 .bind(&pipeline.repository)
1360 .bind(&pipeline.branch)
1361 .bind(&pipeline.workflow_file)
1362 .bind(pipeline.last_run.as_ref().map(|dt| dt.to_rfc3339()))
1363 .bind(Utc::now().to_rfc3339())
1364 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1365 .bind(&pipeline.provider_type)
1366 .bind(resource_state_json(pipeline)?)
1367 .execute(&mut *tx)
1368 .await
1369 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1370 }
1371 }
1372
1373 for old_id in existing.keys() {
1374 if !new_ids.contains_key(old_id) {
1375 sqlx::query(delete_sql)
1376 .bind(old_id)
1377 .execute(&mut *tx)
1378 .await
1379 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1380 }
1381 }
1382
1383 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1384 }
1385 DatabasePool::Postgres(p) => {
1386 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1387
1388 let existing_rows = sqlx::query(select_sql)
1389 .bind(provider_id)
1390 .fetch_all(&mut *tx)
1391 .await
1392 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1393
1394 let existing: HashMap<String, Pipeline> = existing_rows.iter()
1395 .map(|row| {
1396 let pipeline = self.pipeline_from_row(row)?;
1397 Ok((pipeline.id.clone(), pipeline))
1398 })
1399 .collect::<DomainResult<_>>()?;
1400
1401 let new_ids: HashMap<String, &Pipeline> = new_pipelines_clone.iter().map(|p| (p.id.clone(), p)).collect();
1402
1403 for pipeline in &new_pipelines_clone {
1404 let now = Utc::now();
1405 if let Some(old) = existing.get(&pipeline.id) {
1406 if old.status != pipeline.status || old.last_run != pipeline.last_run || old.name != pipeline.name || old.provider_id != provider_id || old.provider_type != pipeline.provider_type
1407 || old.repository != pipeline.repository || old.branch != pipeline.branch || old.workflow_file != pipeline.workflow_file || old.metadata != pipeline.metadata
1408 || old.resource_state != pipeline.resource_state {
1409 sqlx::query(update_sql)
1410 .bind(provider_id).bind(&pipeline.name).bind(pipeline.status.as_str()).bind(&pipeline.repository).bind(&pipeline.branch)
1411 .bind(&pipeline.workflow_file).bind(pipeline.last_run).bind(now)
1412 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1413 .bind(&pipeline.provider_type).bind(resource_state_json(pipeline)?).bind(&pipeline.id)
1414 .execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1415 }
1416 } else {
1417 sqlx::query(insert_sql)
1418 .bind(&pipeline.id).bind(provider_id).bind(&pipeline.name).bind(pipeline.status.as_str()).bind(&pipeline.repository).bind(&pipeline.branch)
1419 .bind(&pipeline.workflow_file).bind(pipeline.last_run).bind(now)
1420 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1421 .bind(&pipeline.provider_type).bind(resource_state_json(pipeline)?)
1422 .execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1423 }
1424 }
1425
1426 for old_id in existing.keys() {
1427 if !new_ids.contains_key(old_id) {
1428 sqlx::query(delete_sql).bind(old_id).execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1429 }
1430 }
1431
1432 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1433 }
1434 }
1435 Ok(())
1436 }
1437 })
1438 .await
1439 }
1440
1441 pub async fn get_pipelines_cache_count(&self) -> DomainResult<i64> {
1442 let count = match &self.cache_pool {
1443 DatabasePool::Sqlite(p) => {
1444 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM pipelines_cache")
1445 .fetch_one(p)
1446 .await
1447 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1448 }
1449 DatabasePool::Postgres(p) => {
1450 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM pipelines_cache")
1451 .fetch_one(p)
1452 .await
1453 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1454 }
1455 };
1456 Ok(count)
1457 }
1458
1459 pub async fn get_run_history_cache_count(&self) -> DomainResult<i64> {
1460 let count = match &self.cache_pool {
1461 DatabasePool::Sqlite(p) => {
1462 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_history_cache")
1463 .fetch_one(p)
1464 .await
1465 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1466 }
1467 DatabasePool::Postgres(p) => {
1468 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_history_cache")
1469 .fetch_one(p)
1470 .await
1471 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1472 }
1473 };
1474 Ok(count)
1475 }
1476
1477 pub async fn get_workflow_params_cache_count(&self) -> DomainResult<i64> {
1478 let count = match &self.cache_pool {
1479 DatabasePool::Sqlite(p) => {
1480 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM workflow_parameters_cache")
1481 .fetch_one(p)
1482 .await
1483 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1484 }
1485 DatabasePool::Postgres(p) => {
1486 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM workflow_parameters_cache")
1487 .fetch_one(p)
1488 .await
1489 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1490 }
1491 };
1492 Ok(count)
1493 }
1494
1495 pub async fn clear_pipelines_cache(&self) -> DomainResult<usize> {
1496 match &self.cache_pool {
1497 DatabasePool::Sqlite(p) => {
1498 let result = sqlx::query("DELETE FROM pipelines_cache")
1499 .execute(p)
1500 .await
1501 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1502 Ok(result.rows_affected() as usize)
1503 }
1504 DatabasePool::Postgres(p) => {
1505 let result = sqlx::query("DELETE FROM pipelines_cache")
1506 .execute(p)
1507 .await
1508 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1509 Ok(result.rows_affected() as usize)
1510 }
1511 }
1512 }
1513
1514 pub async fn clear_all_caches_atomic(&self) -> DomainResult<()> {
1515 match &self.cache_pool {
1516 DatabasePool::Sqlite(p) => {
1517 let mut tx = p
1518 .begin()
1519 .await
1520 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1521
1522 sqlx::query("DELETE FROM run_history_cache")
1523 .execute(&mut *tx)
1524 .await
1525 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1526
1527 sqlx::query("DELETE FROM pipelines_cache")
1528 .execute(&mut *tx)
1529 .await
1530 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1531
1532 sqlx::query("DELETE FROM workflow_parameters_cache")
1533 .execute(&mut *tx)
1534 .await
1535 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1536
1537 tx.commit()
1538 .await
1539 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1540 }
1541 DatabasePool::Postgres(p) => {
1542 let mut tx = p
1543 .begin()
1544 .await
1545 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1546
1547 sqlx::query("DELETE FROM run_history_cache")
1548 .execute(&mut *tx)
1549 .await
1550 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1551
1552 sqlx::query("DELETE FROM pipelines_cache")
1553 .execute(&mut *tx)
1554 .await
1555 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1556
1557 sqlx::query("DELETE FROM workflow_parameters_cache")
1558 .execute(&mut *tx)
1559 .await
1560 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1561
1562 tx.commit()
1563 .await
1564 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1565 }
1566 }
1567
1568 Ok(())
1569 }
1570
1571 pub async fn get_table_preferences(
1572 &self, provider_id: i64, table_id: &str,
1573 ) -> DomainResult<Option<String>> {
1574 self.config_backend
1575 .get_table_preferences(provider_id, table_id)
1576 .await
1577 }
1578
1579 pub async fn upsert_table_preferences(
1580 &self, provider_id: i64, table_id: &str, preferences_json: &str,
1581 ) -> DomainResult<()> {
1582 self.config_backend
1583 .set_table_preferences(provider_id, table_id, preferences_json)
1584 .await
1585 }
1586
1587 pub async fn store_provider_permissions(
1588 &self, provider_id: i64, status: &pipedash_plugin_api::PermissionStatus,
1589 ) -> DomainResult<()> {
1590 let mut permissions_map = HashMap::new();
1591 for perm_check in &status.permissions {
1592 permissions_map.insert(perm_check.permission.name.clone(), perm_check.granted);
1593 }
1594
1595 let stored_permissions = crate::infrastructure::config_backend::StoredPermissions {
1596 permissions: permissions_map,
1597 last_checked: status.checked_at,
1598 };
1599
1600 self.config_backend
1601 .store_permissions(provider_id, &stored_permissions)
1602 .await
1603 }
1604
1605 pub async fn get_provider_permissions(
1606 &self, provider_id: i64,
1607 ) -> DomainResult<Option<pipedash_plugin_api::PermissionStatus>> {
1608 let stored_perms = self.config_backend.get_permissions(provider_id).await?;
1609
1610 match stored_perms {
1611 Some(stored) => {
1612 let mut permissions = Vec::new();
1613 let mut all_granted = true;
1614
1615 for (name, granted) in stored.permissions {
1616 let permission = pipedash_plugin_api::Permission {
1617 name,
1618 description: String::new(),
1619 required: false,
1620 };
1621
1622 permissions.push(pipedash_plugin_api::PermissionCheck {
1623 permission,
1624 granted,
1625 });
1626
1627 if !granted {
1628 all_granted = false;
1629 }
1630 }
1631
1632 Ok(Some(pipedash_plugin_api::PermissionStatus {
1633 permissions,
1634 all_granted,
1635 checked_at: stored.last_checked,
1636 metadata: HashMap::new(),
1637 }))
1638 }
1639 None => Ok(None),
1640 }
1641 }
1642
1643 pub async fn get_cached_run_count(&self, pipeline_id: &str) -> DomainResult<usize> {
1644 let sql = format!(
1645 "SELECT COUNT(*) FROM run_history_cache WHERE pipeline_id = {}",
1646 self.placeholder(1)
1647 );
1648 let count = match &self.cache_pool {
1649 DatabasePool::Sqlite(p) => sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(sql))
1650 .bind(pipeline_id)
1651 .fetch_one(p)
1652 .await
1653 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1654 DatabasePool::Postgres(p) => sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(sql))
1655 .bind(pipeline_id)
1656 .fetch_one(p)
1657 .await
1658 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1659 };
1660
1661 Ok(count as usize)
1662 }
1663
1664 pub async fn get_paginated_runs(
1665 &self, pipeline_id: &str, page: usize, page_size: usize,
1666 ) -> DomainResult<Vec<PipelineRun>> {
1667 let offset = (page - 1) * page_size;
1668
1669 let sql = format!(
1670 "SELECT run_data FROM run_history_cache
1671 WHERE pipeline_id = {}
1672 ORDER BY {}
1673 LIMIT {} OFFSET {}",
1674 self.placeholder(1),
1675 run_order(pipeline_id),
1676 self.placeholder(2),
1677 self.placeholder(3)
1678 );
1679 let runs_json = match &self.cache_pool {
1680 DatabasePool::Sqlite(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
1681 .bind(pipeline_id)
1682 .bind(page_size as i64)
1683 .bind(offset as i64)
1684 .fetch_all(p)
1685 .await
1686 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1687 DatabasePool::Postgres(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
1688 .bind(pipeline_id)
1689 .bind(page_size as i64)
1690 .bind(offset as i64)
1691 .fetch_all(p)
1692 .await
1693 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1694 };
1695
1696 let runs: Vec<PipelineRun> = runs_json
1697 .into_iter()
1698 .filter_map(|json| serde_json::from_str::<PipelineRun>(&json).ok())
1699 .collect();
1700
1701 Ok(runs)
1702 }
1703
1704 fn pipeline_from_row<'r, R>(&self, row: &'r R) -> DomainResult<Pipeline>
1705 where
1706 R: SqlxRow,
1707 usize: sqlx::ColumnIndex<R>,
1708 String: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1709 i64: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1710 Option<String>: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1711 DateTime<Utc>: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1712 Option<DateTime<Utc>>: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1713 {
1714 let id: String = row
1715 .try_get(0)
1716 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1717 let provider_id: i64 = row
1718 .try_get(1)
1719 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1720 let name: String = row
1721 .try_get(2)
1722 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1723 let status_str: String = row
1724 .try_get(3)
1725 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1726 let repository: String = row
1727 .try_get(4)
1728 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1729 let branch: Option<String> = row
1730 .try_get(5)
1731 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1732 let workflow_file: Option<String> = row
1733 .try_get(6)
1734 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1735
1736 let status: PipelineStatus = if status_str.trim().is_empty() {
1737 PipelineStatus::Pending
1738 } else if status_str.starts_with('"') {
1739 serde_json::from_str(&status_str).map_err(|e| {
1740 DomainError::DatabaseError(format!(
1741 "Failed to parse status '{}': {}",
1742 status_str, e
1743 ))
1744 })?
1745 } else {
1746 let quoted = format!("\"{}\"", status_str);
1747 serde_json::from_str("ed).map_err(|e| {
1748 DomainError::DatabaseError(format!(
1749 "Failed to parse status '{}': {}",
1750 status_str, e
1751 ))
1752 })?
1753 };
1754
1755 let (last_run, last_updated) = match &self.cache_pool {
1756 DatabasePool::Sqlite(_) => {
1757 let last_run_str: Option<String> = row
1758 .try_get(7)
1759 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1760 let last_updated_str: String = row
1761 .try_get(8)
1762 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1763
1764 let last_run = last_run_str
1765 .and_then(|s| DateTime::parse_from_rfc3339(&s).ok())
1766 .map(|dt| dt.with_timezone(&Utc));
1767 let last_updated = DateTime::parse_from_rfc3339(&last_updated_str)
1768 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1769 .with_timezone(&Utc);
1770
1771 (last_run, last_updated)
1772 }
1773 DatabasePool::Postgres(_) => {
1774 let last_run: Option<DateTime<Utc>> = row
1775 .try_get(7)
1776 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1777 let last_updated: DateTime<Utc> = row
1778 .try_get(8)
1779 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1780
1781 (last_run, last_updated)
1782 }
1783 };
1784
1785 let provider_type: String = row
1786 .try_get(9)
1787 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1788 let metadata_json: String = row
1789 .try_get(10)
1790 .map_err(|error| DomainError::DatabaseError(error.to_string()))?;
1791 let metadata = serde_json::from_str(&metadata_json).map_err(|error| {
1792 DomainError::DatabaseError(format!("Invalid pipeline metadata: {error}"))
1793 })?;
1794 let resource_state = row
1795 .try_get::<Option<String>, _>(11)
1796 .map_err(|error| DomainError::DatabaseError(error.to_string()))?
1797 .map(|json| serde_json::from_str(&json))
1798 .transpose()
1799 .map_err(|error| {
1800 DomainError::DatabaseError(format!("Invalid pipeline resource state: {error}"))
1801 })?;
1802
1803 Ok(Pipeline {
1804 id,
1805 provider_id,
1806 provider_type,
1807 name,
1808 status,
1809 last_run,
1810 last_updated,
1811 repository,
1812 branch,
1813 workflow_file,
1814 resource_state,
1815 metadata,
1816 overdue_run: None,
1817 typical_duration_seconds: None,
1818 })
1819 }
1820}
1821
1822fn resource_state_json(pipeline: &Pipeline) -> DomainResult<Option<String>> {
1823 pipeline
1824 .resource_state
1825 .as_ref()
1826 .map(serde_json::to_string)
1827 .transpose()
1828 .map_err(|error| DomainError::DatabaseError(error.to_string()))
1829}
1830
1831#[cfg(test)]
1832mod tests {
1833 use serde_json::json;
1834
1835 use super::*;
1836 use crate::domain::ResourceState;
1837 use crate::infrastructure::{
1838 MemoryTokenStore,
1839 PipedashConfig,
1840 StorageManager,
1841 };
1842
1843 async fn test_repository(dir: &tempfile::TempDir) -> (StorageManager, Repository) {
1844 let mut config = PipedashConfig::default();
1845 config.storage.data_dir = dir.path().display().to_string();
1846 let manager =
1847 StorageManager::with_token_store(config, Arc::new(MemoryTokenStore::new()), true)
1848 .await
1849 .unwrap();
1850 let repository = Repository::new(
1851 manager.config_backend(),
1852 manager.database_pool(),
1853 manager.token_store().await,
1854 );
1855
1856 (manager, repository)
1857 }
1858
1859 fn cached_run(pipeline_id: &str, run_number: i64, started_at: &str) -> PipelineRun {
1860 serde_json::from_value(json!({
1861 "id": format!("{pipeline_id}__{run_number}"), "pipeline_id": pipeline_id,
1862 "run_number": run_number, "status": "success", "started_at": started_at,
1863 "logs_url": "", "metadata": {}
1864 }))
1865 .unwrap()
1866 }
1867
1868 async fn run_pages(repository: &Repository, pipeline_id: &str) -> Vec<Vec<i64>> {
1869 let mut pages = Vec::new();
1870 for page in 1..=3 {
1871 let runs = repository
1872 .get_paginated_runs(pipeline_id, page, 2)
1873 .await
1874 .unwrap();
1875 pages.push(runs.iter().map(|run| run.run_number).collect());
1876 }
1877
1878 pages
1879 }
1880
1881 #[tokio::test]
1882 async fn pipeline_cache_keeps_resource_state_and_stores_state_only_changes() {
1883 let dir = tempfile::tempdir().unwrap();
1884 let (manager, repository) = test_repository(&dir).await;
1885 let provider_id = manager
1886 .config_backend()
1887 .create_provider(&ProviderConfig {
1888 id: None,
1889 name: "Flux".to_string(),
1890 provider_type: "fluxcd".to_string(),
1891 token: String::new(),
1892 config: HashMap::new(),
1893 refresh_interval: 30,
1894 version: None,
1895 })
1896 .await
1897 .unwrap();
1898 let mut pipeline: Pipeline = serde_json::from_value(json!({
1899 "id": "fluxcd__1__Kustomization__apps__web", "provider_id": provider_id,
1900 "provider_type": "fluxcd", "name": "Kustomization/web", "status": "running",
1901 "repository": "apps/source", "last_updated": "2026-10-21T00:00:00Z",
1902 "resource_state": { "label": "Reconciling", "message": "Applying revision a" },
1903 "metadata": { "uid": "resource-1" }
1904 }))
1905 .unwrap();
1906 let mut stateless = pipeline.clone();
1907 stateless.id = "fluxcd__1__Kustomization__apps__api".to_string();
1908 stateless.resource_state = None;
1909
1910 repository
1911 .update_pipelines_cache(provider_id, &[pipeline.clone(), stateless])
1912 .await
1913 .unwrap();
1914 let first = repository
1915 .get_cached_pipelines(Some(provider_id))
1916 .await
1917 .unwrap();
1918 let stored = |pipelines: &[Pipeline], id: &str| {
1919 pipelines
1920 .iter()
1921 .find(|candidate| candidate.id == id)
1922 .cloned()
1923 .unwrap()
1924 };
1925 assert_eq!(
1926 stored(&first, &pipeline.id).resource_state,
1927 pipeline.resource_state
1928 );
1929 assert_eq!(stored(&first, &pipeline.id).metadata, pipeline.metadata);
1930 assert_eq!(
1931 stored(&first, "fluxcd__1__Kustomization__apps__api").resource_state,
1932 None
1933 );
1934
1935 pipeline.resource_state = Some(ResourceState::new(
1936 "Reconciling",
1937 Some("Applying revision b"),
1938 ));
1939 repository
1940 .update_pipelines_cache(provider_id, std::slice::from_ref(&pipeline))
1941 .await
1942 .unwrap();
1943 let changed = repository.get_cached_pipelines(None).await.unwrap();
1944 assert_eq!(changed.len(), 1);
1945 assert_eq!(changed[0].resource_state, pipeline.resource_state);
1946 assert_eq!(changed[0].status, pipeline.status);
1947 }
1948
1949 fn jenkins_provider() -> ProviderConfig {
1950 ProviderConfig {
1951 id: None,
1952 name: "ci-jenkins".to_string(),
1953 provider_type: "jenkins".to_string(),
1954 token: String::new(),
1955 config: HashMap::new(),
1956 refresh_interval: 60,
1957 version: None,
1958 }
1959 }
1960
1961 #[tokio::test]
1962 async fn get_provider_reads_a_missing_token_entry_as_empty() {
1963 let dir = tempfile::tempdir().unwrap();
1964 let (manager, repository) = test_repository(&dir).await;
1965 let id = manager
1966 .config_backend()
1967 .create_provider(&jenkins_provider())
1968 .await
1969 .unwrap();
1970
1971 let provider = repository.get_provider(id).await.unwrap();
1972
1973 assert_eq!(provider.name, "ci-jenkins");
1974 assert_eq!(provider.token, "");
1975 }
1976
1977 struct FailingTokenStore;
1978
1979 #[async_trait::async_trait]
1980 impl TokenStore for FailingTokenStore {
1981 async fn store_token(&self, _provider_id: i64, _token: &str) -> DomainResult<()> {
1982 Ok(())
1983 }
1984
1985 async fn get_token(&self, _provider_id: i64) -> DomainResult<String> {
1986 Err(DomainError::AuthenticationFailed(
1987 "vault is locked".to_string(),
1988 ))
1989 }
1990
1991 async fn delete_token(&self, _provider_id: i64) -> DomainResult<()> {
1992 Ok(())
1993 }
1994
1995 async fn get_all_tokens(&self) -> DomainResult<HashMap<i64, String>> {
1996 Ok(HashMap::new())
1997 }
1998 }
1999
2000 #[tokio::test]
2001 async fn get_provider_fails_when_the_token_store_fails() {
2002 let dir = tempfile::tempdir().unwrap();
2003 let (manager, _) = test_repository(&dir).await;
2004 let id = manager
2005 .config_backend()
2006 .create_provider(&jenkins_provider())
2007 .await
2008 .unwrap();
2009 let repository = Repository::new(
2010 manager.config_backend(),
2011 manager.database_pool(),
2012 Arc::new(FailingTokenStore),
2013 );
2014
2015 let result = repository.get_provider(id).await;
2016
2017 assert!(matches!(result, Err(DomainError::AuthenticationFailed(_))));
2018 }
2019
2020 #[tokio::test]
2021 async fn native_run_cache_lists_the_latest_activity_first_on_every_page() {
2022 let dir = tempfile::tempdir().unwrap();
2023 let (_manager, repository) = test_repository(&dir).await;
2024 let flux = "fluxcd__1__Kustomization__apps__web";
2025 let other = "github__1__acme__web";
2026 let runs = |pipeline_id: &str| {
2027 [
2028 cached_run(pipeline_id, 100, "2026-10-22T10:00:00Z"),
2029 cached_run(pipeline_id, 200, "2026-10-20T10:00:00Z"),
2030 cached_run(pipeline_id, 300, "2026-10-21T10:00:00.500Z"),
2031 cached_run(pipeline_id, 400, "2026-10-21T10:00:00Z"),
2032 cached_run(pipeline_id, 500, "2026-10-21T10:00:00Z"),
2033 ]
2034 };
2035 repository
2036 .cache_run_history(flux, &runs(flux))
2037 .await
2038 .unwrap();
2039 repository
2040 .cache_run_history(other, &runs(other))
2041 .await
2042 .unwrap();
2043
2044 assert_eq!(
2045 run_pages(&repository, flux).await,
2046 vec![vec![100, 300], vec![500, 400], vec![200]]
2047 );
2048 assert_eq!(
2049 run_pages(&repository, other).await,
2050 vec![vec![500, 400], vec![300, 200], vec![100]]
2051 );
2052
2053 let mut reconciled_again = runs(flux)[1].clone();
2054 reconciled_again.started_at = "2026-10-23T10:00:00Z".parse().unwrap();
2055 let mut archived = runs(flux)[0].clone();
2056 archived
2057 .metadata
2058 .insert("archived".to_string(), json!(true));
2059 repository
2060 .merge_run_cache(
2061 flux,
2062 Vec::new(),
2063 vec![reconciled_again, archived],
2064 Vec::new(),
2065 )
2066 .await
2067 .unwrap();
2068
2069 assert_eq!(
2070 run_pages(&repository, flux).await,
2071 vec![vec![200, 100], vec![300, 500], vec![400]]
2072 );
2073 let newest = repository.get_cached_run_history(flux, 3).await.unwrap();
2074 assert_eq!(
2075 newest.iter().map(|run| run.run_number).collect::<Vec<_>>(),
2076 vec![200, 100, 300]
2077 );
2078 let kept = repository.get_cached_run(flux, 100).await.unwrap().unwrap();
2079 assert_eq!(kept.metadata["archived"], json!(true));
2080 assert_eq!(repository.get_cached_run_count(flux).await.unwrap(), 5);
2081 }
2082
2083 #[tokio::test]
2084 async fn sqlite_migration_backfills_the_order_key_of_cached_flux_runs() {
2085 let pool = sqlx::sqlite::SqlitePoolOptions::new()
2086 .max_connections(1)
2087 .connect("sqlite::memory:")
2088 .await
2089 .unwrap();
2090 sqlx::raw_sql(
2091 "CREATE TABLE run_history_cache (
2092 pipeline_id TEXT NOT NULL,
2093 run_number INTEGER NOT NULL,
2094 run_data TEXT NOT NULL,
2095 PRIMARY KEY (pipeline_id, run_number)
2096 )",
2097 )
2098 .execute(&pool)
2099 .await
2100 .unwrap();
2101 let timestamps = [
2102 "2026-10-21T10:00:00Z",
2103 "2026-10-21T10:00:00.1Z",
2104 "2026-10-21T10:00:00.500999Z",
2105 "2026-10-21T10:00:00.999999Z",
2106 ];
2107 let pipelines = [
2108 "fluxcd__1__Kustomization__apps__web",
2109 "github__1__acme__web",
2110 ];
2111 for pipeline_id in pipelines {
2112 for (index, timestamp) in timestamps.iter().enumerate() {
2113 sqlx::query("INSERT INTO run_history_cache (pipeline_id, run_number, run_data) VALUES (?, ?, ?)")
2114 .bind(pipeline_id)
2115 .bind((index + 1) as i64)
2116 .bind(json!({ "started_at": timestamp }).to_string())
2117 .execute(&pool).await.unwrap();
2118 }
2119 }
2120
2121 sqlx::raw_sql(include_str!(
2122 "../../../migrations/sqlite/20261021000001_run_history_started_at.sql"
2123 ))
2124 .execute(&pool)
2125 .await
2126 .unwrap();
2127
2128 let keys: Vec<(String, i64, i64)> = sqlx::query_as(
2129 "SELECT pipeline_id, run_number, started_at_ms FROM run_history_cache ORDER BY pipeline_id, run_number",
2130 ).fetch_all(&pool).await.unwrap();
2131 let expected: Vec<_> = pipelines
2132 .iter()
2133 .flat_map(|pipeline_id| {
2134 timestamps
2135 .iter()
2136 .enumerate()
2137 .map(move |(index, timestamp)| {
2138 let millis = if pipeline_id.starts_with("fluxcd__") {
2139 timestamp
2140 .parse::<DateTime<Utc>>()
2141 .unwrap()
2142 .timestamp_millis()
2143 } else {
2144 0
2145 };
2146 (pipeline_id.to_string(), (index + 1) as i64, millis)
2147 })
2148 })
2149 .collect();
2150 assert_eq!(keys, expected);
2151 }
2152}