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 async fn get_cached_run_history(
898 &self, pipeline_id: &str, limit: usize,
899 ) -> DomainResult<Vec<PipelineRun>> {
900 let sql = format!(
901 "SELECT run_data FROM run_history_cache
902 WHERE pipeline_id = {}
903 ORDER BY {}
904 LIMIT {}",
905 self.placeholder(1),
906 run_order(pipeline_id),
907 self.placeholder(2)
908 );
909 let rows = match &self.cache_pool {
910 DatabasePool::Sqlite(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
911 .bind(pipeline_id)
912 .bind(limit as i64)
913 .fetch_all(p)
914 .await
915 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
916 DatabasePool::Postgres(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
917 .bind(pipeline_id)
918 .bind(limit as i64)
919 .fetch_all(p)
920 .await
921 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
922 };
923
924 let runs: Vec<PipelineRun> = rows
925 .iter()
926 .filter_map(|json| serde_json::from_str(json).ok())
927 .collect();
928
929 Ok(runs)
930 }
931
932 pub async fn get_cached_run(
933 &self, pipeline_id: &str, run_number: i64,
934 ) -> DomainResult<Option<PipelineRun>> {
935 let sql = format!(
936 "SELECT run_data FROM run_history_cache WHERE pipeline_id = {} AND run_number = {}",
937 self.placeholder(1),
938 self.placeholder(2)
939 );
940 let row = match &self.cache_pool {
941 DatabasePool::Sqlite(pool) => {
942 sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
943 .bind(pipeline_id)
944 .bind(run_number)
945 .fetch_optional(pool)
946 .await
947 }
948 DatabasePool::Postgres(pool) => {
949 sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
950 .bind(pipeline_id)
951 .bind(run_number)
952 .fetch_optional(pool)
953 .await
954 }
955 }
956 .map_err(|error| DomainError::DatabaseError(error.to_string()))?;
957
958 row.map(|json| serde_json::from_str(&json))
959 .transpose()
960 .map_err(|error| DomainError::DatabaseError(format!("Invalid cached run: {error}")))
961 }
962
963 pub async fn clear_cached_run_history(&self, pipeline_id: &str) -> DomainResult<()> {
964 let sql = format!(
965 "DELETE FROM run_history_cache WHERE pipeline_id = {}",
966 self.placeholder(1)
967 );
968 match &self.cache_pool {
969 DatabasePool::Sqlite(p) => {
970 sqlx::query(sqlx::AssertSqlSafe(sql))
971 .bind(pipeline_id)
972 .execute(p)
973 .await
974 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
975 }
976 DatabasePool::Postgres(p) => {
977 sqlx::query(sqlx::AssertSqlSafe(sql))
978 .bind(pipeline_id)
979 .execute(p)
980 .await
981 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
982 }
983 }
984
985 Ok(())
986 }
987
988 pub async fn clear_all_run_history_cache(&self) -> DomainResult<()> {
989 match &self.cache_pool {
990 DatabasePool::Sqlite(p) => {
991 sqlx::query("DELETE FROM run_history_cache")
992 .execute(p)
993 .await
994 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
995 }
996 DatabasePool::Postgres(p) => {
997 sqlx::query("DELETE FROM run_history_cache")
998 .execute(p)
999 .await
1000 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1001 }
1002 }
1003
1004 Ok(())
1005 }
1006
1007 pub async fn get_cached_runs_with_hashes(
1008 &self, pipeline_id: &str,
1009 ) -> DomainResult<HashMap<i64, (PipelineRun, String)>> {
1010 let mut result = HashMap::new();
1011 match &self.cache_pool {
1012 DatabasePool::Sqlite(p) => {
1013 let sql = format!(
1014 "SELECT run_number, run_data, run_hash FROM run_history_cache WHERE pipeline_id = {}",
1015 self.placeholder(1)
1016 );
1017 let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
1018 .bind(pipeline_id)
1019 .fetch_all(p)
1020 .await
1021 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1022
1023 for row in rows {
1024 let run_number: i64 = row
1025 .try_get(0)
1026 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1027 let run_data: String = row
1028 .try_get(1)
1029 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1030 let run_hash: String = row
1031 .try_get(2)
1032 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1033
1034 if let Ok(run) = serde_json::from_str::<PipelineRun>(&run_data) {
1035 result.insert(run_number, (run, run_hash));
1036 }
1037 }
1038 }
1039 DatabasePool::Postgres(p) => {
1040 let sql = format!(
1041 "SELECT run_number::BIGINT, run_data, run_hash FROM run_history_cache WHERE pipeline_id = {}",
1042 self.placeholder(1)
1043 );
1044 let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
1045 .bind(pipeline_id)
1046 .fetch_all(p)
1047 .await
1048 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1049
1050 for row in rows {
1051 let run_number: i64 = row
1052 .try_get(0)
1053 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1054 let run_data: String = row
1055 .try_get(1)
1056 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1057 let run_hash: String = row
1058 .try_get(2)
1059 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1060
1061 if let Ok(run) = serde_json::from_str::<PipelineRun>(&run_data) {
1062 result.insert(run_number, (run, run_hash));
1063 }
1064 }
1065 }
1066 }
1067
1068 Ok(result)
1069 }
1070
1071 pub async fn merge_run_cache(
1072 &self, pipeline_id: &str, new_runs: Vec<PipelineRun>, changed_runs: Vec<PipelineRun>,
1073 deleted_run_numbers: Vec<i64>,
1074 ) -> DomainResult<()> {
1075 let start = std::time::Instant::now();
1076 let pipeline_id_str = pipeline_id.to_string();
1077 let pool = self.cache_pool.clone();
1078
1079 let new_runs_count = new_runs.len();
1080 let changed_runs_count = changed_runs.len();
1081 let deleted_runs_count = deleted_run_numbers.len();
1082
1083 retry_on_busy(move || {
1084 let pipeline_id_clone = pipeline_id_str.clone();
1085 let new_runs_clone = new_runs.clone();
1086 let changed_runs_clone = changed_runs.clone();
1087 let deleted_clone = deleted_run_numbers.clone();
1088 let pool_clone = pool.clone();
1089
1090 async move {
1091 const BATCH_SIZE: usize = 100;
1092 const DELETE_BATCH_SIZE: usize = 100;
1093
1094 let all_runs: Vec<_> = new_runs_clone.iter().chain(changed_runs_clone.iter()).collect();
1095
1096 match &pool_clone {
1097 DatabasePool::Sqlite(p) => {
1098 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1099
1100 if !all_runs.is_empty() {
1101 for chunk in all_runs.chunks(BATCH_SIZE) {
1102 let prepared_data = chunk
1103 .iter()
1104 .copied()
1105 .map(run_cache_row)
1106 .collect::<DomainResult<Vec<_>>>()?;
1107
1108 let values_clause = prepared_data
1109 .iter()
1110 .map(|_| "(?, ?, ?, datetime('now'), ?, ?)")
1111 .collect::<Vec<_>>()
1112 .join(", ");
1113
1114 let sql = format!(
1115 "INSERT OR REPLACE INTO run_history_cache (pipeline_id, run_number, run_data, fetched_at, run_hash, started_at_ms) VALUES {}",
1116 values_clause
1117 );
1118
1119 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
1120 for (run_number, run_data, run_hash, started_at_ms) in &prepared_data {
1121 query = query
1122 .bind(&pipeline_id_clone)
1123 .bind(run_number)
1124 .bind(run_data)
1125 .bind(run_hash)
1126 .bind(started_at_ms);
1127 }
1128
1129 query.execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1130 }
1131 }
1132
1133 if !deleted_clone.is_empty() {
1134 for chunk in deleted_clone.chunks(DELETE_BATCH_SIZE) {
1135 let placeholders = chunk.iter().map(|_| "?").collect::<Vec<_>>().join(", ");
1136 let sql = format!(
1137 "DELETE FROM run_history_cache WHERE pipeline_id = ? AND run_number IN ({})",
1138 placeholders
1139 );
1140
1141 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(&pipeline_id_clone);
1142 for run_number in chunk {
1143 query = query.bind(run_number);
1144 }
1145
1146 query.execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1147 }
1148 }
1149
1150 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1151 }
1152 DatabasePool::Postgres(p) => {
1153 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1154
1155 if !all_runs.is_empty() {
1156 for chunk in all_runs.chunks(BATCH_SIZE) {
1157 let prepared_data = chunk
1158 .iter()
1159 .copied()
1160 .map(run_cache_row)
1161 .collect::<DomainResult<Vec<_>>>()?;
1162
1163 let mut param_idx = 0;
1164 let values_clauses: Vec<String> = prepared_data
1165 .iter()
1166 .map(|_| {
1167 let clause = format!(
1168 "(${}, ${}, ${}, NOW(), ${}, ${})",
1169 param_idx + 1, param_idx + 2, param_idx + 3, param_idx + 4, param_idx + 5
1170 );
1171 param_idx += 5;
1172 clause
1173 })
1174 .collect();
1175
1176 let sql = format!(
1177 "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",
1178 values_clauses.join(", ")
1179 );
1180
1181 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
1182 for (run_number, run_data, run_hash, started_at_ms) in &prepared_data {
1183 query = query
1184 .bind(&pipeline_id_clone)
1185 .bind(run_number)
1186 .bind(run_data)
1187 .bind(run_hash)
1188 .bind(started_at_ms);
1189 }
1190
1191 query.execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1192 }
1193 }
1194
1195 if !deleted_clone.is_empty() {
1196 for chunk in deleted_clone.chunks(DELETE_BATCH_SIZE) {
1197 let mut param_idx = 1;
1198 let placeholders = chunk.iter().map(|_| {
1199 let p = format!("${}", param_idx);
1200 param_idx += 1;
1201 p
1202 }).collect::<Vec<_>>().join(", ");
1203
1204 let sql = format!(
1205 "DELETE FROM run_history_cache WHERE pipeline_id = ${} AND run_number IN ({})",
1206 param_idx,
1207 placeholders
1208 );
1209
1210 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
1211 for run_number in chunk {
1212 query = query.bind(run_number);
1213 }
1214 query = query.bind(&pipeline_id_clone);
1215
1216 query.execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1217 }
1218 }
1219
1220 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1221 }
1222 }
1223
1224 Ok(())
1225 }
1226 })
1227 .await?;
1228
1229 let elapsed = start.elapsed();
1230 tracing::debug!(
1231 pipeline_id = pipeline_id,
1232 new_runs = new_runs_count,
1233 changed_runs = changed_runs_count,
1234 deleted_runs = deleted_runs_count,
1235 elapsed_ms = elapsed.as_millis(),
1236 "Merged run cache (batch operations)"
1237 );
1238
1239 Ok(())
1240 }
1241
1242 pub async fn update_pipelines_cache(
1243 &self, provider_id: i64, new_pipelines: &[Pipeline],
1244 ) -> DomainResult<()> {
1245 let new_pipelines_vec = new_pipelines.to_vec();
1246 let pool = self.cache_pool.clone();
1247 let is_postgres = pool.is_postgres();
1248
1249 retry_on_busy(move || {
1250 let new_pipelines_clone = new_pipelines_vec.clone();
1251 let pool_clone = pool.clone();
1252 let is_pg = is_postgres;
1253 async move {
1254 let select_sql = if is_pg {
1255 "SELECT id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, provider_type, metadata_json, resource_state_json
1256 FROM pipelines_cache
1257 WHERE provider_id = $1"
1258 } else {
1259 "SELECT id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, provider_type, metadata_json, resource_state_json
1260 FROM pipelines_cache
1261 WHERE provider_id = ?"
1262 };
1263
1264 let update_sql = if is_pg {
1265 "UPDATE pipelines_cache
1266 SET provider_id = $1, name = $2, status = $3, repository = $4, branch = $5,
1267 workflow_file = $6, last_run = $7, last_updated = $8, metadata_json = $9, provider_type = $10,
1268 resource_state_json = $11
1269 WHERE id = $12"
1270 } else {
1271 "UPDATE pipelines_cache
1272 SET provider_id = ?, name = ?, status = ?, repository = ?, branch = ?,
1273 workflow_file = ?, last_run = ?, last_updated = ?, metadata_json = ?, provider_type = ?,
1274 resource_state_json = ?
1275 WHERE id = ?"
1276 };
1277
1278 let insert_sql = if is_pg {
1279 "INSERT INTO pipelines_cache
1280 (id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, metadata_json, provider_type, resource_state_json)
1281 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)"
1282 } else {
1283 "INSERT INTO pipelines_cache
1284 (id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, metadata_json, provider_type, resource_state_json)
1285 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
1286 };
1287
1288 let delete_sql = if is_pg {
1289 "DELETE FROM pipelines_cache WHERE id = $1"
1290 } else {
1291 "DELETE FROM pipelines_cache WHERE id = ?"
1292 };
1293
1294 match &pool_clone {
1295 DatabasePool::Sqlite(p) => {
1296 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1297
1298 let existing_rows = sqlx::query(select_sql)
1299 .bind(provider_id)
1300 .fetch_all(&mut *tx)
1301 .await
1302 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1303
1304 let existing: HashMap<String, Pipeline> = existing_rows.iter()
1305 .map(|row| {
1306 let pipeline = self.pipeline_from_row(row)?;
1307 Ok((pipeline.id.clone(), pipeline))
1308 })
1309 .collect::<DomainResult<_>>()?;
1310
1311 let new_ids: HashMap<String, &Pipeline> =
1312 new_pipelines_clone.iter().map(|p| (p.id.clone(), p)).collect();
1313
1314 for pipeline in &new_pipelines_clone {
1315 if let Some(old) = existing.get(&pipeline.id) {
1316 if old.status != pipeline.status
1317 || old.last_run != pipeline.last_run
1318 || old.name != pipeline.name
1319 || old.provider_id != provider_id
1320 || old.provider_type != pipeline.provider_type
1321 || old.repository != pipeline.repository
1322 || old.branch != pipeline.branch
1323 || old.workflow_file != pipeline.workflow_file
1324 || old.metadata != pipeline.metadata
1325 || old.resource_state != pipeline.resource_state
1326 {
1327 sqlx::query(update_sql)
1328 .bind(provider_id)
1329 .bind(&pipeline.name)
1330 .bind(pipeline.status.as_str())
1331 .bind(&pipeline.repository)
1332 .bind(&pipeline.branch)
1333 .bind(&pipeline.workflow_file)
1334 .bind(pipeline.last_run.as_ref().map(|dt| dt.to_rfc3339()))
1335 .bind(Utc::now().to_rfc3339())
1336 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1337 .bind(&pipeline.provider_type)
1338 .bind(resource_state_json(pipeline)?)
1339 .bind(&pipeline.id)
1340 .execute(&mut *tx)
1341 .await
1342 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1343 }
1344 } else {
1345 sqlx::query(insert_sql)
1346 .bind(&pipeline.id)
1347 .bind(provider_id)
1348 .bind(&pipeline.name)
1349 .bind(pipeline.status.as_str())
1350 .bind(&pipeline.repository)
1351 .bind(&pipeline.branch)
1352 .bind(&pipeline.workflow_file)
1353 .bind(pipeline.last_run.as_ref().map(|dt| dt.to_rfc3339()))
1354 .bind(Utc::now().to_rfc3339())
1355 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1356 .bind(&pipeline.provider_type)
1357 .bind(resource_state_json(pipeline)?)
1358 .execute(&mut *tx)
1359 .await
1360 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1361 }
1362 }
1363
1364 for old_id in existing.keys() {
1365 if !new_ids.contains_key(old_id) {
1366 sqlx::query(delete_sql)
1367 .bind(old_id)
1368 .execute(&mut *tx)
1369 .await
1370 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1371 }
1372 }
1373
1374 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1375 }
1376 DatabasePool::Postgres(p) => {
1377 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1378
1379 let existing_rows = sqlx::query(select_sql)
1380 .bind(provider_id)
1381 .fetch_all(&mut *tx)
1382 .await
1383 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1384
1385 let existing: HashMap<String, Pipeline> = existing_rows.iter()
1386 .map(|row| {
1387 let pipeline = self.pipeline_from_row(row)?;
1388 Ok((pipeline.id.clone(), pipeline))
1389 })
1390 .collect::<DomainResult<_>>()?;
1391
1392 let new_ids: HashMap<String, &Pipeline> = new_pipelines_clone.iter().map(|p| (p.id.clone(), p)).collect();
1393
1394 for pipeline in &new_pipelines_clone {
1395 let now = Utc::now();
1396 if let Some(old) = existing.get(&pipeline.id) {
1397 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
1398 || old.repository != pipeline.repository || old.branch != pipeline.branch || old.workflow_file != pipeline.workflow_file || old.metadata != pipeline.metadata
1399 || old.resource_state != pipeline.resource_state {
1400 sqlx::query(update_sql)
1401 .bind(provider_id).bind(&pipeline.name).bind(pipeline.status.as_str()).bind(&pipeline.repository).bind(&pipeline.branch)
1402 .bind(&pipeline.workflow_file).bind(pipeline.last_run).bind(now)
1403 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1404 .bind(&pipeline.provider_type).bind(resource_state_json(pipeline)?).bind(&pipeline.id)
1405 .execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1406 }
1407 } else {
1408 sqlx::query(insert_sql)
1409 .bind(&pipeline.id).bind(provider_id).bind(&pipeline.name).bind(pipeline.status.as_str()).bind(&pipeline.repository).bind(&pipeline.branch)
1410 .bind(&pipeline.workflow_file).bind(pipeline.last_run).bind(now)
1411 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1412 .bind(&pipeline.provider_type).bind(resource_state_json(pipeline)?)
1413 .execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1414 }
1415 }
1416
1417 for old_id in existing.keys() {
1418 if !new_ids.contains_key(old_id) {
1419 sqlx::query(delete_sql).bind(old_id).execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1420 }
1421 }
1422
1423 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1424 }
1425 }
1426 Ok(())
1427 }
1428 })
1429 .await
1430 }
1431
1432 pub async fn get_pipelines_cache_count(&self) -> DomainResult<i64> {
1433 let count = match &self.cache_pool {
1434 DatabasePool::Sqlite(p) => {
1435 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM pipelines_cache")
1436 .fetch_one(p)
1437 .await
1438 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1439 }
1440 DatabasePool::Postgres(p) => {
1441 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM pipelines_cache")
1442 .fetch_one(p)
1443 .await
1444 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1445 }
1446 };
1447 Ok(count)
1448 }
1449
1450 pub async fn get_run_history_cache_count(&self) -> DomainResult<i64> {
1451 let count = match &self.cache_pool {
1452 DatabasePool::Sqlite(p) => {
1453 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_history_cache")
1454 .fetch_one(p)
1455 .await
1456 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1457 }
1458 DatabasePool::Postgres(p) => {
1459 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_history_cache")
1460 .fetch_one(p)
1461 .await
1462 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1463 }
1464 };
1465 Ok(count)
1466 }
1467
1468 pub async fn get_workflow_params_cache_count(&self) -> DomainResult<i64> {
1469 let count = match &self.cache_pool {
1470 DatabasePool::Sqlite(p) => {
1471 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM workflow_parameters_cache")
1472 .fetch_one(p)
1473 .await
1474 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1475 }
1476 DatabasePool::Postgres(p) => {
1477 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM workflow_parameters_cache")
1478 .fetch_one(p)
1479 .await
1480 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1481 }
1482 };
1483 Ok(count)
1484 }
1485
1486 pub async fn clear_pipelines_cache(&self) -> DomainResult<usize> {
1487 match &self.cache_pool {
1488 DatabasePool::Sqlite(p) => {
1489 let result = sqlx::query("DELETE FROM pipelines_cache")
1490 .execute(p)
1491 .await
1492 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1493 Ok(result.rows_affected() as usize)
1494 }
1495 DatabasePool::Postgres(p) => {
1496 let result = sqlx::query("DELETE FROM pipelines_cache")
1497 .execute(p)
1498 .await
1499 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1500 Ok(result.rows_affected() as usize)
1501 }
1502 }
1503 }
1504
1505 pub async fn clear_all_caches_atomic(&self) -> DomainResult<()> {
1506 match &self.cache_pool {
1507 DatabasePool::Sqlite(p) => {
1508 let mut tx = p
1509 .begin()
1510 .await
1511 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1512
1513 sqlx::query("DELETE FROM run_history_cache")
1514 .execute(&mut *tx)
1515 .await
1516 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1517
1518 sqlx::query("DELETE FROM pipelines_cache")
1519 .execute(&mut *tx)
1520 .await
1521 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1522
1523 sqlx::query("DELETE FROM workflow_parameters_cache")
1524 .execute(&mut *tx)
1525 .await
1526 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1527
1528 tx.commit()
1529 .await
1530 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1531 }
1532 DatabasePool::Postgres(p) => {
1533 let mut tx = p
1534 .begin()
1535 .await
1536 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1537
1538 sqlx::query("DELETE FROM run_history_cache")
1539 .execute(&mut *tx)
1540 .await
1541 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1542
1543 sqlx::query("DELETE FROM pipelines_cache")
1544 .execute(&mut *tx)
1545 .await
1546 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1547
1548 sqlx::query("DELETE FROM workflow_parameters_cache")
1549 .execute(&mut *tx)
1550 .await
1551 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1552
1553 tx.commit()
1554 .await
1555 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1556 }
1557 }
1558
1559 Ok(())
1560 }
1561
1562 pub async fn get_table_preferences(
1563 &self, provider_id: i64, table_id: &str,
1564 ) -> DomainResult<Option<String>> {
1565 self.config_backend
1566 .get_table_preferences(provider_id, table_id)
1567 .await
1568 }
1569
1570 pub async fn upsert_table_preferences(
1571 &self, provider_id: i64, table_id: &str, preferences_json: &str,
1572 ) -> DomainResult<()> {
1573 self.config_backend
1574 .set_table_preferences(provider_id, table_id, preferences_json)
1575 .await
1576 }
1577
1578 pub async fn store_provider_permissions(
1579 &self, provider_id: i64, status: &pipedash_plugin_api::PermissionStatus,
1580 ) -> DomainResult<()> {
1581 let mut permissions_map = HashMap::new();
1582 for perm_check in &status.permissions {
1583 permissions_map.insert(perm_check.permission.name.clone(), perm_check.granted);
1584 }
1585
1586 let stored_permissions = crate::infrastructure::config_backend::StoredPermissions {
1587 permissions: permissions_map,
1588 last_checked: status.checked_at,
1589 };
1590
1591 self.config_backend
1592 .store_permissions(provider_id, &stored_permissions)
1593 .await
1594 }
1595
1596 pub async fn get_provider_permissions(
1597 &self, provider_id: i64,
1598 ) -> DomainResult<Option<pipedash_plugin_api::PermissionStatus>> {
1599 let stored_perms = self.config_backend.get_permissions(provider_id).await?;
1600
1601 match stored_perms {
1602 Some(stored) => {
1603 let mut permissions = Vec::new();
1604 let mut all_granted = true;
1605
1606 for (name, granted) in stored.permissions {
1607 let permission = pipedash_plugin_api::Permission {
1608 name,
1609 description: String::new(),
1610 required: false,
1611 };
1612
1613 permissions.push(pipedash_plugin_api::PermissionCheck {
1614 permission,
1615 granted,
1616 });
1617
1618 if !granted {
1619 all_granted = false;
1620 }
1621 }
1622
1623 Ok(Some(pipedash_plugin_api::PermissionStatus {
1624 permissions,
1625 all_granted,
1626 checked_at: stored.last_checked,
1627 metadata: HashMap::new(),
1628 }))
1629 }
1630 None => Ok(None),
1631 }
1632 }
1633
1634 pub async fn get_cached_run_count(&self, pipeline_id: &str) -> DomainResult<usize> {
1635 let sql = format!(
1636 "SELECT COUNT(*) FROM run_history_cache WHERE pipeline_id = {}",
1637 self.placeholder(1)
1638 );
1639 let count = match &self.cache_pool {
1640 DatabasePool::Sqlite(p) => sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(sql))
1641 .bind(pipeline_id)
1642 .fetch_one(p)
1643 .await
1644 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1645 DatabasePool::Postgres(p) => sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(sql))
1646 .bind(pipeline_id)
1647 .fetch_one(p)
1648 .await
1649 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1650 };
1651
1652 Ok(count as usize)
1653 }
1654
1655 pub async fn get_paginated_runs(
1656 &self, pipeline_id: &str, page: usize, page_size: usize,
1657 ) -> DomainResult<Vec<PipelineRun>> {
1658 let offset = (page - 1) * page_size;
1659
1660 let sql = format!(
1661 "SELECT run_data FROM run_history_cache
1662 WHERE pipeline_id = {}
1663 ORDER BY {}
1664 LIMIT {} OFFSET {}",
1665 self.placeholder(1),
1666 run_order(pipeline_id),
1667 self.placeholder(2),
1668 self.placeholder(3)
1669 );
1670 let runs_json = match &self.cache_pool {
1671 DatabasePool::Sqlite(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
1672 .bind(pipeline_id)
1673 .bind(page_size as i64)
1674 .bind(offset as i64)
1675 .fetch_all(p)
1676 .await
1677 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1678 DatabasePool::Postgres(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
1679 .bind(pipeline_id)
1680 .bind(page_size as i64)
1681 .bind(offset as i64)
1682 .fetch_all(p)
1683 .await
1684 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1685 };
1686
1687 let runs: Vec<PipelineRun> = runs_json
1688 .into_iter()
1689 .filter_map(|json| serde_json::from_str::<PipelineRun>(&json).ok())
1690 .collect();
1691
1692 Ok(runs)
1693 }
1694
1695 fn pipeline_from_row<'r, R>(&self, row: &'r R) -> DomainResult<Pipeline>
1696 where
1697 R: SqlxRow,
1698 usize: sqlx::ColumnIndex<R>,
1699 String: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1700 i64: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1701 Option<String>: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1702 DateTime<Utc>: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1703 Option<DateTime<Utc>>: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1704 {
1705 let id: String = row
1706 .try_get(0)
1707 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1708 let provider_id: i64 = row
1709 .try_get(1)
1710 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1711 let name: String = row
1712 .try_get(2)
1713 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1714 let status_str: String = row
1715 .try_get(3)
1716 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1717 let repository: String = row
1718 .try_get(4)
1719 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1720 let branch: Option<String> = row
1721 .try_get(5)
1722 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1723 let workflow_file: Option<String> = row
1724 .try_get(6)
1725 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1726
1727 let status: PipelineStatus = if status_str.trim().is_empty() {
1728 PipelineStatus::Pending
1729 } else if status_str.starts_with('"') {
1730 serde_json::from_str(&status_str).map_err(|e| {
1731 DomainError::DatabaseError(format!(
1732 "Failed to parse status '{}': {}",
1733 status_str, e
1734 ))
1735 })?
1736 } else {
1737 let quoted = format!("\"{}\"", status_str);
1738 serde_json::from_str("ed).map_err(|e| {
1739 DomainError::DatabaseError(format!(
1740 "Failed to parse status '{}': {}",
1741 status_str, e
1742 ))
1743 })?
1744 };
1745
1746 let (last_run, last_updated) = match &self.cache_pool {
1747 DatabasePool::Sqlite(_) => {
1748 let last_run_str: Option<String> = row
1749 .try_get(7)
1750 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1751 let last_updated_str: String = row
1752 .try_get(8)
1753 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1754
1755 let last_run = last_run_str
1756 .and_then(|s| DateTime::parse_from_rfc3339(&s).ok())
1757 .map(|dt| dt.with_timezone(&Utc));
1758 let last_updated = DateTime::parse_from_rfc3339(&last_updated_str)
1759 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1760 .with_timezone(&Utc);
1761
1762 (last_run, last_updated)
1763 }
1764 DatabasePool::Postgres(_) => {
1765 let last_run: Option<DateTime<Utc>> = row
1766 .try_get(7)
1767 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1768 let last_updated: DateTime<Utc> = row
1769 .try_get(8)
1770 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1771
1772 (last_run, last_updated)
1773 }
1774 };
1775
1776 let provider_type: String = row
1777 .try_get(9)
1778 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1779 let metadata_json: String = row
1780 .try_get(10)
1781 .map_err(|error| DomainError::DatabaseError(error.to_string()))?;
1782 let metadata = serde_json::from_str(&metadata_json).map_err(|error| {
1783 DomainError::DatabaseError(format!("Invalid pipeline metadata: {error}"))
1784 })?;
1785 let resource_state = row
1786 .try_get::<Option<String>, _>(11)
1787 .map_err(|error| DomainError::DatabaseError(error.to_string()))?
1788 .map(|json| serde_json::from_str(&json))
1789 .transpose()
1790 .map_err(|error| {
1791 DomainError::DatabaseError(format!("Invalid pipeline resource state: {error}"))
1792 })?;
1793
1794 Ok(Pipeline {
1795 id,
1796 provider_id,
1797 provider_type,
1798 name,
1799 status,
1800 last_run,
1801 last_updated,
1802 repository,
1803 branch,
1804 workflow_file,
1805 resource_state,
1806 metadata,
1807 overdue_run: None,
1808 typical_duration_seconds: None,
1809 })
1810 }
1811}
1812
1813fn resource_state_json(pipeline: &Pipeline) -> DomainResult<Option<String>> {
1814 pipeline
1815 .resource_state
1816 .as_ref()
1817 .map(serde_json::to_string)
1818 .transpose()
1819 .map_err(|error| DomainError::DatabaseError(error.to_string()))
1820}
1821
1822#[cfg(test)]
1823mod tests {
1824 use serde_json::json;
1825
1826 use super::*;
1827 use crate::domain::ResourceState;
1828 use crate::infrastructure::{
1829 MemoryTokenStore,
1830 PipedashConfig,
1831 StorageManager,
1832 };
1833
1834 async fn test_repository(dir: &tempfile::TempDir) -> (StorageManager, Repository) {
1835 let mut config = PipedashConfig::default();
1836 config.storage.data_dir = dir.path().display().to_string();
1837 let manager =
1838 StorageManager::with_token_store(config, Arc::new(MemoryTokenStore::new()), true)
1839 .await
1840 .unwrap();
1841 let repository = Repository::new(
1842 manager.config_backend(),
1843 manager.database_pool(),
1844 manager.token_store().await,
1845 );
1846
1847 (manager, repository)
1848 }
1849
1850 fn cached_run(pipeline_id: &str, run_number: i64, started_at: &str) -> PipelineRun {
1851 serde_json::from_value(json!({
1852 "id": format!("{pipeline_id}__{run_number}"), "pipeline_id": pipeline_id,
1853 "run_number": run_number, "status": "success", "started_at": started_at,
1854 "logs_url": "", "metadata": {}
1855 }))
1856 .unwrap()
1857 }
1858
1859 async fn run_pages(repository: &Repository, pipeline_id: &str) -> Vec<Vec<i64>> {
1860 let mut pages = Vec::new();
1861 for page in 1..=3 {
1862 let runs = repository
1863 .get_paginated_runs(pipeline_id, page, 2)
1864 .await
1865 .unwrap();
1866 pages.push(runs.iter().map(|run| run.run_number).collect());
1867 }
1868
1869 pages
1870 }
1871
1872 #[tokio::test]
1873 async fn pipeline_cache_keeps_resource_state_and_stores_state_only_changes() {
1874 let dir = tempfile::tempdir().unwrap();
1875 let (manager, repository) = test_repository(&dir).await;
1876 let provider_id = manager
1877 .config_backend()
1878 .create_provider(&ProviderConfig {
1879 id: None,
1880 name: "Flux".to_string(),
1881 provider_type: "fluxcd".to_string(),
1882 token: String::new(),
1883 config: HashMap::new(),
1884 refresh_interval: 30,
1885 version: None,
1886 })
1887 .await
1888 .unwrap();
1889 let mut pipeline: Pipeline = serde_json::from_value(json!({
1890 "id": "fluxcd__1__Kustomization__apps__web", "provider_id": provider_id,
1891 "provider_type": "fluxcd", "name": "Kustomization/web", "status": "running",
1892 "repository": "apps/source", "last_updated": "2026-10-21T00:00:00Z",
1893 "resource_state": { "label": "Reconciling", "message": "Applying revision a" },
1894 "metadata": { "uid": "resource-1" }
1895 }))
1896 .unwrap();
1897 let mut stateless = pipeline.clone();
1898 stateless.id = "fluxcd__1__Kustomization__apps__api".to_string();
1899 stateless.resource_state = None;
1900
1901 repository
1902 .update_pipelines_cache(provider_id, &[pipeline.clone(), stateless])
1903 .await
1904 .unwrap();
1905 let first = repository
1906 .get_cached_pipelines(Some(provider_id))
1907 .await
1908 .unwrap();
1909 let stored = |pipelines: &[Pipeline], id: &str| {
1910 pipelines
1911 .iter()
1912 .find(|candidate| candidate.id == id)
1913 .cloned()
1914 .unwrap()
1915 };
1916 assert_eq!(
1917 stored(&first, &pipeline.id).resource_state,
1918 pipeline.resource_state
1919 );
1920 assert_eq!(stored(&first, &pipeline.id).metadata, pipeline.metadata);
1921 assert_eq!(
1922 stored(&first, "fluxcd__1__Kustomization__apps__api").resource_state,
1923 None
1924 );
1925
1926 pipeline.resource_state = Some(ResourceState::new(
1927 "Reconciling",
1928 Some("Applying revision b"),
1929 ));
1930 repository
1931 .update_pipelines_cache(provider_id, std::slice::from_ref(&pipeline))
1932 .await
1933 .unwrap();
1934 let changed = repository.get_cached_pipelines(None).await.unwrap();
1935 assert_eq!(changed.len(), 1);
1936 assert_eq!(changed[0].resource_state, pipeline.resource_state);
1937 assert_eq!(changed[0].status, pipeline.status);
1938 }
1939
1940 fn jenkins_provider() -> ProviderConfig {
1941 ProviderConfig {
1942 id: None,
1943 name: "ci-jenkins".to_string(),
1944 provider_type: "jenkins".to_string(),
1945 token: String::new(),
1946 config: HashMap::new(),
1947 refresh_interval: 60,
1948 version: None,
1949 }
1950 }
1951
1952 #[tokio::test]
1953 async fn get_provider_reads_a_missing_token_entry_as_empty() {
1954 let dir = tempfile::tempdir().unwrap();
1955 let (manager, repository) = test_repository(&dir).await;
1956 let id = manager
1957 .config_backend()
1958 .create_provider(&jenkins_provider())
1959 .await
1960 .unwrap();
1961
1962 let provider = repository.get_provider(id).await.unwrap();
1963
1964 assert_eq!(provider.name, "ci-jenkins");
1965 assert_eq!(provider.token, "");
1966 }
1967
1968 struct FailingTokenStore;
1969
1970 #[async_trait::async_trait]
1971 impl TokenStore for FailingTokenStore {
1972 async fn store_token(&self, _provider_id: i64, _token: &str) -> DomainResult<()> {
1973 Ok(())
1974 }
1975
1976 async fn get_token(&self, _provider_id: i64) -> DomainResult<String> {
1977 Err(DomainError::AuthenticationFailed(
1978 "vault is locked".to_string(),
1979 ))
1980 }
1981
1982 async fn delete_token(&self, _provider_id: i64) -> DomainResult<()> {
1983 Ok(())
1984 }
1985
1986 async fn get_all_tokens(&self) -> DomainResult<HashMap<i64, String>> {
1987 Ok(HashMap::new())
1988 }
1989 }
1990
1991 #[tokio::test]
1992 async fn get_provider_fails_when_the_token_store_fails() {
1993 let dir = tempfile::tempdir().unwrap();
1994 let (manager, _) = test_repository(&dir).await;
1995 let id = manager
1996 .config_backend()
1997 .create_provider(&jenkins_provider())
1998 .await
1999 .unwrap();
2000 let repository = Repository::new(
2001 manager.config_backend(),
2002 manager.database_pool(),
2003 Arc::new(FailingTokenStore),
2004 );
2005
2006 let result = repository.get_provider(id).await;
2007
2008 assert!(matches!(result, Err(DomainError::AuthenticationFailed(_))));
2009 }
2010
2011 #[tokio::test]
2012 async fn native_run_cache_lists_the_latest_activity_first_on_every_page() {
2013 let dir = tempfile::tempdir().unwrap();
2014 let (_manager, repository) = test_repository(&dir).await;
2015 let flux = "fluxcd__1__Kustomization__apps__web";
2016 let other = "github__1__acme__web";
2017 let runs = |pipeline_id: &str| {
2018 [
2019 cached_run(pipeline_id, 100, "2026-10-22T10:00:00Z"),
2020 cached_run(pipeline_id, 200, "2026-10-20T10:00:00Z"),
2021 cached_run(pipeline_id, 300, "2026-10-21T10:00:00.500Z"),
2022 cached_run(pipeline_id, 400, "2026-10-21T10:00:00Z"),
2023 cached_run(pipeline_id, 500, "2026-10-21T10:00:00Z"),
2024 ]
2025 };
2026 repository
2027 .cache_run_history(flux, &runs(flux))
2028 .await
2029 .unwrap();
2030 repository
2031 .cache_run_history(other, &runs(other))
2032 .await
2033 .unwrap();
2034
2035 assert_eq!(
2036 run_pages(&repository, flux).await,
2037 vec![vec![100, 300], vec![500, 400], vec![200]]
2038 );
2039 assert_eq!(
2040 run_pages(&repository, other).await,
2041 vec![vec![500, 400], vec![300, 200], vec![100]]
2042 );
2043
2044 let mut reconciled_again = runs(flux)[1].clone();
2045 reconciled_again.started_at = "2026-10-23T10:00:00Z".parse().unwrap();
2046 let mut archived = runs(flux)[0].clone();
2047 archived
2048 .metadata
2049 .insert("archived".to_string(), json!(true));
2050 repository
2051 .merge_run_cache(
2052 flux,
2053 Vec::new(),
2054 vec![reconciled_again, archived],
2055 Vec::new(),
2056 )
2057 .await
2058 .unwrap();
2059
2060 assert_eq!(
2061 run_pages(&repository, flux).await,
2062 vec![vec![200, 100], vec![300, 500], vec![400]]
2063 );
2064 let newest = repository.get_cached_run_history(flux, 3).await.unwrap();
2065 assert_eq!(
2066 newest.iter().map(|run| run.run_number).collect::<Vec<_>>(),
2067 vec![200, 100, 300]
2068 );
2069 let kept = repository.get_cached_run(flux, 100).await.unwrap().unwrap();
2070 assert_eq!(kept.metadata["archived"], json!(true));
2071 assert_eq!(repository.get_cached_run_count(flux).await.unwrap(), 5);
2072 }
2073
2074 #[tokio::test]
2075 async fn sqlite_migration_backfills_the_order_key_of_cached_flux_runs() {
2076 let pool = sqlx::sqlite::SqlitePoolOptions::new()
2077 .max_connections(1)
2078 .connect("sqlite::memory:")
2079 .await
2080 .unwrap();
2081 sqlx::raw_sql(
2082 "CREATE TABLE run_history_cache (
2083 pipeline_id TEXT NOT NULL,
2084 run_number INTEGER NOT NULL,
2085 run_data TEXT NOT NULL,
2086 PRIMARY KEY (pipeline_id, run_number)
2087 )",
2088 )
2089 .execute(&pool)
2090 .await
2091 .unwrap();
2092 let timestamps = [
2093 "2026-10-21T10:00:00Z",
2094 "2026-10-21T10:00:00.1Z",
2095 "2026-10-21T10:00:00.500999Z",
2096 "2026-10-21T10:00:00.999999Z",
2097 ];
2098 let pipelines = [
2099 "fluxcd__1__Kustomization__apps__web",
2100 "github__1__acme__web",
2101 ];
2102 for pipeline_id in pipelines {
2103 for (index, timestamp) in timestamps.iter().enumerate() {
2104 sqlx::query("INSERT INTO run_history_cache (pipeline_id, run_number, run_data) VALUES (?, ?, ?)")
2105 .bind(pipeline_id)
2106 .bind((index + 1) as i64)
2107 .bind(json!({ "started_at": timestamp }).to_string())
2108 .execute(&pool).await.unwrap();
2109 }
2110 }
2111
2112 sqlx::raw_sql(include_str!(
2113 "../../../migrations/sqlite/20261021000001_run_history_started_at.sql"
2114 ))
2115 .execute(&pool)
2116 .await
2117 .unwrap();
2118
2119 let keys: Vec<(String, i64, i64)> = sqlx::query_as(
2120 "SELECT pipeline_id, run_number, started_at_ms FROM run_history_cache ORDER BY pipeline_id, run_number",
2121 ).fetch_all(&pool).await.unwrap();
2122 let expected: Vec<_> = pipelines
2123 .iter()
2124 .flat_map(|pipeline_id| {
2125 timestamps
2126 .iter()
2127 .enumerate()
2128 .map(move |(index, timestamp)| {
2129 let millis = if pipeline_id.starts_with("fluxcd__") {
2130 timestamp
2131 .parse::<DateTime<Utc>>()
2132 .unwrap()
2133 .timestamp_millis()
2134 } else {
2135 0
2136 };
2137 (pipeline_id.to_string(), (index + 1) as i64, millis)
2138 })
2139 })
2140 .collect();
2141 assert_eq!(keys, expected);
2142 }
2143}