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 .delete_provider_members(id)
410 .await
411 {
412 tracing::warn!("Failed to remove pipeline view members of provider {id}: {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
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
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
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
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 WHERE id = $11"
1269 } else {
1270 "UPDATE pipelines_cache
1271 SET provider_id = ?, name = ?, status = ?, repository = ?, branch = ?,
1272 workflow_file = ?, last_run = ?, last_updated = ?, metadata_json = ?, provider_type = ?
1273 WHERE id = ?"
1274 };
1275
1276 let insert_sql = if is_pg {
1277 "INSERT INTO pipelines_cache
1278 (id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, metadata_json, provider_type)
1279 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)"
1280 } else {
1281 "INSERT INTO pipelines_cache
1282 (id, provider_id, name, status, repository, branch, workflow_file, last_run, last_updated, metadata_json, provider_type)
1283 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
1284 };
1285
1286 let delete_sql = if is_pg {
1287 "DELETE FROM pipelines_cache WHERE id = $1"
1288 } else {
1289 "DELETE FROM pipelines_cache WHERE id = ?"
1290 };
1291
1292 match &pool_clone {
1293 DatabasePool::Sqlite(p) => {
1294 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1295
1296 let existing_rows = sqlx::query(select_sql)
1297 .bind(provider_id)
1298 .fetch_all(&mut *tx)
1299 .await
1300 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1301
1302 let existing: HashMap<String, Pipeline> = existing_rows.iter()
1303 .map(|row| {
1304 let pipeline = self.pipeline_from_row(row)?;
1305 Ok((pipeline.id.clone(), pipeline))
1306 })
1307 .collect::<DomainResult<_>>()?;
1308
1309 let new_ids: HashMap<String, &Pipeline> =
1310 new_pipelines_clone.iter().map(|p| (p.id.clone(), p)).collect();
1311
1312 for pipeline in &new_pipelines_clone {
1313 if let Some(old) = existing.get(&pipeline.id) {
1314 if old.status != pipeline.status
1315 || old.last_run != pipeline.last_run
1316 || old.name != pipeline.name
1317 || old.provider_id != provider_id
1318 || old.provider_type != pipeline.provider_type
1319 || old.repository != pipeline.repository
1320 || old.branch != pipeline.branch
1321 || old.workflow_file != pipeline.workflow_file
1322 || old.metadata != pipeline.metadata
1323 {
1324 sqlx::query(update_sql)
1325 .bind(provider_id)
1326 .bind(&pipeline.name)
1327 .bind(pipeline.status.as_str())
1328 .bind(&pipeline.repository)
1329 .bind(&pipeline.branch)
1330 .bind(&pipeline.workflow_file)
1331 .bind(pipeline.last_run.as_ref().map(|dt| dt.to_rfc3339()))
1332 .bind(Utc::now().to_rfc3339())
1333 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1334 .bind(&pipeline.provider_type)
1335 .bind(&pipeline.id)
1336 .execute(&mut *tx)
1337 .await
1338 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1339 }
1340 } else {
1341 sqlx::query(insert_sql)
1342 .bind(&pipeline.id)
1343 .bind(provider_id)
1344 .bind(&pipeline.name)
1345 .bind(pipeline.status.as_str())
1346 .bind(&pipeline.repository)
1347 .bind(&pipeline.branch)
1348 .bind(&pipeline.workflow_file)
1349 .bind(pipeline.last_run.as_ref().map(|dt| dt.to_rfc3339()))
1350 .bind(Utc::now().to_rfc3339())
1351 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1352 .bind(&pipeline.provider_type)
1353 .execute(&mut *tx)
1354 .await
1355 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1356 }
1357 }
1358
1359 for old_id in existing.keys() {
1360 if !new_ids.contains_key(old_id) {
1361 sqlx::query(delete_sql)
1362 .bind(old_id)
1363 .execute(&mut *tx)
1364 .await
1365 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1366 }
1367 }
1368
1369 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1370 }
1371 DatabasePool::Postgres(p) => {
1372 let mut tx = p.begin().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1373
1374 let existing_rows = sqlx::query(select_sql)
1375 .bind(provider_id)
1376 .fetch_all(&mut *tx)
1377 .await
1378 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1379
1380 let existing: HashMap<String, Pipeline> = existing_rows.iter()
1381 .map(|row| {
1382 let pipeline = self.pipeline_from_row(row)?;
1383 Ok((pipeline.id.clone(), pipeline))
1384 })
1385 .collect::<DomainResult<_>>()?;
1386
1387 let new_ids: HashMap<String, &Pipeline> = new_pipelines_clone.iter().map(|p| (p.id.clone(), p)).collect();
1388
1389 for pipeline in &new_pipelines_clone {
1390 let now = Utc::now();
1391 if let Some(old) = existing.get(&pipeline.id) {
1392 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
1393 || old.repository != pipeline.repository || old.branch != pipeline.branch || old.workflow_file != pipeline.workflow_file || old.metadata != pipeline.metadata {
1394 sqlx::query(update_sql)
1395 .bind(provider_id).bind(&pipeline.name).bind(pipeline.status.as_str()).bind(&pipeline.repository).bind(&pipeline.branch)
1396 .bind(&pipeline.workflow_file).bind(pipeline.last_run).bind(now)
1397 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1398 .bind(&pipeline.provider_type).bind(&pipeline.id)
1399 .execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1400 }
1401 } else {
1402 sqlx::query(insert_sql)
1403 .bind(&pipeline.id).bind(provider_id).bind(&pipeline.name).bind(pipeline.status.as_str()).bind(&pipeline.repository).bind(&pipeline.branch)
1404 .bind(&pipeline.workflow_file).bind(pipeline.last_run).bind(now)
1405 .bind(serde_json::to_string(&pipeline.metadata).map_err(|e| DomainError::DatabaseError(e.to_string()))?)
1406 .bind(&pipeline.provider_type)
1407 .execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1408 }
1409 }
1410
1411 for old_id in existing.keys() {
1412 if !new_ids.contains_key(old_id) {
1413 sqlx::query(delete_sql).bind(old_id).execute(&mut *tx).await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1414 }
1415 }
1416
1417 tx.commit().await.map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1418 }
1419 }
1420 Ok(())
1421 }
1422 })
1423 .await
1424 }
1425
1426 pub async fn get_pipelines_cache_count(&self) -> DomainResult<i64> {
1427 let count = match &self.cache_pool {
1428 DatabasePool::Sqlite(p) => {
1429 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM pipelines_cache")
1430 .fetch_one(p)
1431 .await
1432 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1433 }
1434 DatabasePool::Postgres(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 };
1441 Ok(count)
1442 }
1443
1444 pub async fn get_run_history_cache_count(&self) -> DomainResult<i64> {
1445 let count = match &self.cache_pool {
1446 DatabasePool::Sqlite(p) => {
1447 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM run_history_cache")
1448 .fetch_one(p)
1449 .await
1450 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1451 }
1452 DatabasePool::Postgres(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 };
1459 Ok(count)
1460 }
1461
1462 pub async fn get_workflow_params_cache_count(&self) -> DomainResult<i64> {
1463 let count = match &self.cache_pool {
1464 DatabasePool::Sqlite(p) => {
1465 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM workflow_parameters_cache")
1466 .fetch_one(p)
1467 .await
1468 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1469 }
1470 DatabasePool::Postgres(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 };
1477 Ok(count)
1478 }
1479
1480 pub async fn clear_pipelines_cache(&self) -> DomainResult<usize> {
1481 match &self.cache_pool {
1482 DatabasePool::Sqlite(p) => {
1483 let result = sqlx::query("DELETE FROM pipelines_cache")
1484 .execute(p)
1485 .await
1486 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1487 Ok(result.rows_affected() as usize)
1488 }
1489 DatabasePool::Postgres(p) => {
1490 let result = sqlx::query("DELETE FROM pipelines_cache")
1491 .execute(p)
1492 .await
1493 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1494 Ok(result.rows_affected() as usize)
1495 }
1496 }
1497 }
1498
1499 pub async fn clear_all_caches_atomic(&self) -> DomainResult<()> {
1500 match &self.cache_pool {
1501 DatabasePool::Sqlite(p) => {
1502 let mut tx = p
1503 .begin()
1504 .await
1505 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1506
1507 sqlx::query("DELETE FROM run_history_cache")
1508 .execute(&mut *tx)
1509 .await
1510 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1511
1512 sqlx::query("DELETE FROM pipelines_cache")
1513 .execute(&mut *tx)
1514 .await
1515 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1516
1517 sqlx::query("DELETE FROM workflow_parameters_cache")
1518 .execute(&mut *tx)
1519 .await
1520 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1521
1522 tx.commit()
1523 .await
1524 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1525 }
1526 DatabasePool::Postgres(p) => {
1527 let mut tx = p
1528 .begin()
1529 .await
1530 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1531
1532 sqlx::query("DELETE FROM run_history_cache")
1533 .execute(&mut *tx)
1534 .await
1535 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1536
1537 sqlx::query("DELETE FROM pipelines_cache")
1538 .execute(&mut *tx)
1539 .await
1540 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1541
1542 sqlx::query("DELETE FROM workflow_parameters_cache")
1543 .execute(&mut *tx)
1544 .await
1545 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1546
1547 tx.commit()
1548 .await
1549 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1550 }
1551 }
1552
1553 Ok(())
1554 }
1555
1556 pub async fn get_table_preferences(
1557 &self, provider_id: i64, table_id: &str,
1558 ) -> DomainResult<Option<String>> {
1559 self.config_backend
1560 .get_table_preferences(provider_id, table_id)
1561 .await
1562 }
1563
1564 pub async fn upsert_table_preferences(
1565 &self, provider_id: i64, table_id: &str, preferences_json: &str,
1566 ) -> DomainResult<()> {
1567 self.config_backend
1568 .set_table_preferences(provider_id, table_id, preferences_json)
1569 .await
1570 }
1571
1572 pub async fn store_provider_permissions(
1573 &self, provider_id: i64, status: &pipedash_plugin_api::PermissionStatus,
1574 ) -> DomainResult<()> {
1575 let mut permissions_map = HashMap::new();
1576 for perm_check in &status.permissions {
1577 permissions_map.insert(perm_check.permission.name.clone(), perm_check.granted);
1578 }
1579
1580 let stored_permissions = crate::infrastructure::config_backend::StoredPermissions {
1581 permissions: permissions_map,
1582 last_checked: status.checked_at,
1583 };
1584
1585 self.config_backend
1586 .store_permissions(provider_id, &stored_permissions)
1587 .await
1588 }
1589
1590 pub async fn get_provider_permissions(
1591 &self, provider_id: i64,
1592 ) -> DomainResult<Option<pipedash_plugin_api::PermissionStatus>> {
1593 let stored_perms = self.config_backend.get_permissions(provider_id).await?;
1594
1595 match stored_perms {
1596 Some(stored) => {
1597 let mut permissions = Vec::new();
1598 let mut all_granted = true;
1599
1600 for (name, granted) in stored.permissions {
1601 let permission = pipedash_plugin_api::Permission {
1602 name,
1603 description: String::new(),
1604 required: false,
1605 };
1606
1607 permissions.push(pipedash_plugin_api::PermissionCheck {
1608 permission,
1609 granted,
1610 });
1611
1612 if !granted {
1613 all_granted = false;
1614 }
1615 }
1616
1617 Ok(Some(pipedash_plugin_api::PermissionStatus {
1618 permissions,
1619 all_granted,
1620 checked_at: stored.last_checked,
1621 metadata: HashMap::new(),
1622 }))
1623 }
1624 None => Ok(None),
1625 }
1626 }
1627
1628 pub async fn get_cached_run_count(&self, pipeline_id: &str) -> DomainResult<usize> {
1629 let sql = format!(
1630 "SELECT COUNT(*) FROM run_history_cache WHERE pipeline_id = {}",
1631 self.placeholder(1)
1632 );
1633 let count = match &self.cache_pool {
1634 DatabasePool::Sqlite(p) => sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(sql))
1635 .bind(pipeline_id)
1636 .fetch_one(p)
1637 .await
1638 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1639 DatabasePool::Postgres(p) => sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(sql))
1640 .bind(pipeline_id)
1641 .fetch_one(p)
1642 .await
1643 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1644 };
1645
1646 Ok(count as usize)
1647 }
1648
1649 pub async fn get_paginated_runs(
1650 &self, pipeline_id: &str, page: usize, page_size: usize,
1651 ) -> DomainResult<Vec<PipelineRun>> {
1652 let offset = (page - 1) * page_size;
1653
1654 let sql = format!(
1655 "SELECT run_data FROM run_history_cache
1656 WHERE pipeline_id = {}
1657 ORDER BY {}
1658 LIMIT {} OFFSET {}",
1659 self.placeholder(1),
1660 run_order(pipeline_id),
1661 self.placeholder(2),
1662 self.placeholder(3)
1663 );
1664 let runs_json = match &self.cache_pool {
1665 DatabasePool::Sqlite(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
1666 .bind(pipeline_id)
1667 .bind(page_size as i64)
1668 .bind(offset as i64)
1669 .fetch_all(p)
1670 .await
1671 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1672 DatabasePool::Postgres(p) => sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql))
1673 .bind(pipeline_id)
1674 .bind(page_size as i64)
1675 .bind(offset as i64)
1676 .fetch_all(p)
1677 .await
1678 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1679 };
1680
1681 let runs: Vec<PipelineRun> = runs_json
1682 .into_iter()
1683 .filter_map(|json| serde_json::from_str::<PipelineRun>(&json).ok())
1684 .collect();
1685
1686 Ok(runs)
1687 }
1688
1689 fn pipeline_from_row<'r, R>(&self, row: &'r R) -> DomainResult<Pipeline>
1690 where
1691 R: SqlxRow,
1692 usize: sqlx::ColumnIndex<R>,
1693 String: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1694 i64: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1695 Option<String>: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1696 DateTime<Utc>: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1697 Option<DateTime<Utc>>: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
1698 {
1699 let id: String = row
1700 .try_get(0)
1701 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1702 let provider_id: i64 = row
1703 .try_get(1)
1704 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1705 let name: String = row
1706 .try_get(2)
1707 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1708 let status_str: String = row
1709 .try_get(3)
1710 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1711 let repository: String = row
1712 .try_get(4)
1713 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1714 let branch: Option<String> = row
1715 .try_get(5)
1716 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1717 let workflow_file: Option<String> = row
1718 .try_get(6)
1719 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1720
1721 let status: PipelineStatus = if status_str.trim().is_empty() {
1722 PipelineStatus::Pending
1723 } else if status_str.starts_with('"') {
1724 serde_json::from_str(&status_str).map_err(|e| {
1725 DomainError::DatabaseError(format!(
1726 "Failed to parse status '{}': {}",
1727 status_str, e
1728 ))
1729 })?
1730 } else {
1731 let quoted = format!("\"{}\"", status_str);
1732 serde_json::from_str("ed).map_err(|e| {
1733 DomainError::DatabaseError(format!(
1734 "Failed to parse status '{}': {}",
1735 status_str, e
1736 ))
1737 })?
1738 };
1739
1740 let (last_run, last_updated) = match &self.cache_pool {
1741 DatabasePool::Sqlite(_) => {
1742 let last_run_str: Option<String> = row
1743 .try_get(7)
1744 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1745 let last_updated_str: String = row
1746 .try_get(8)
1747 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1748
1749 let last_run = last_run_str
1750 .and_then(|s| DateTime::parse_from_rfc3339(&s).ok())
1751 .map(|dt| dt.with_timezone(&Utc));
1752 let last_updated = DateTime::parse_from_rfc3339(&last_updated_str)
1753 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1754 .with_timezone(&Utc);
1755
1756 (last_run, last_updated)
1757 }
1758 DatabasePool::Postgres(_) => {
1759 let last_run: Option<DateTime<Utc>> = row
1760 .try_get(7)
1761 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1762 let last_updated: DateTime<Utc> = row
1763 .try_get(8)
1764 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1765
1766 (last_run, last_updated)
1767 }
1768 };
1769
1770 let provider_type: String = row
1771 .try_get(9)
1772 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1773 let metadata_json: String = row
1774 .try_get(10)
1775 .map_err(|error| DomainError::DatabaseError(error.to_string()))?;
1776 let metadata = serde_json::from_str(&metadata_json).map_err(|error| {
1777 DomainError::DatabaseError(format!("Invalid pipeline metadata: {error}"))
1778 })?;
1779
1780 Ok(Pipeline {
1781 id,
1782 provider_id,
1783 provider_type,
1784 name,
1785 status,
1786 last_run,
1787 last_updated,
1788 repository,
1789 branch,
1790 workflow_file,
1791 metadata,
1792 overdue_run: None,
1793 })
1794 }
1795}
1796
1797#[cfg(test)]
1798mod tests {
1799 use serde_json::json;
1800
1801 use super::*;
1802 use crate::infrastructure::{
1803 MemoryTokenStore,
1804 PipedashConfig,
1805 StorageManager,
1806 };
1807
1808 async fn test_repository(dir: &tempfile::TempDir) -> (StorageManager, Repository) {
1809 let mut config = PipedashConfig::default();
1810 config.storage.data_dir = dir.path().display().to_string();
1811 let manager =
1812 StorageManager::with_token_store(config, Arc::new(MemoryTokenStore::new()), true)
1813 .await
1814 .unwrap();
1815 let repository = Repository::new(
1816 manager.config_backend(),
1817 manager.database_pool(),
1818 manager.token_store().await,
1819 );
1820
1821 (manager, repository)
1822 }
1823
1824 fn cached_run(pipeline_id: &str, run_number: i64, started_at: &str) -> PipelineRun {
1825 serde_json::from_value(json!({
1826 "id": format!("{pipeline_id}__{run_number}"), "pipeline_id": pipeline_id,
1827 "run_number": run_number, "status": "success", "started_at": started_at,
1828 "logs_url": "", "metadata": {}
1829 }))
1830 .unwrap()
1831 }
1832
1833 async fn run_pages(repository: &Repository, pipeline_id: &str) -> Vec<Vec<i64>> {
1834 let mut pages = Vec::new();
1835 for page in 1..=3 {
1836 let runs = repository
1837 .get_paginated_runs(pipeline_id, page, 2)
1838 .await
1839 .unwrap();
1840 pages.push(runs.iter().map(|run| run.run_number).collect());
1841 }
1842
1843 pages
1844 }
1845
1846 #[tokio::test]
1847 async fn pipeline_cache_preserves_metadata_and_metadata_only_updates() {
1848 let dir = tempfile::tempdir().unwrap();
1849 let (manager, repository) = test_repository(&dir).await;
1850 let provider_id = manager
1851 .config_backend()
1852 .create_provider(&ProviderConfig {
1853 id: None,
1854 name: "Flux".to_string(),
1855 provider_type: "fluxcd".to_string(),
1856 token: String::new(),
1857 config: HashMap::new(),
1858 refresh_interval: 30,
1859 version: None,
1860 })
1861 .await
1862 .unwrap();
1863 let mut pipeline: Pipeline = serde_json::from_value(json!({
1864 "id": "fluxcd__1__Kustomization__apps__web", "provider_id": provider_id,
1865 "provider_type": "fluxcd", "name": "Kustomization/web", "status": "success",
1866 "repository": "apps/source", "last_updated": "2026-10-21T00:00:00Z",
1867 "metadata": { "uid": "resource-1", "status_label": "Ready", "message": "Applied revision a" }
1868 })).unwrap();
1869
1870 repository
1871 .update_pipelines_cache(provider_id, std::slice::from_ref(&pipeline))
1872 .await
1873 .unwrap();
1874 let first = repository
1875 .get_cached_pipelines(Some(provider_id))
1876 .await
1877 .unwrap();
1878 assert_eq!(first[0].metadata, pipeline.metadata);
1879
1880 pipeline
1881 .metadata
1882 .insert("message".to_string(), json!("Applied revision b"));
1883 repository
1884 .update_pipelines_cache(provider_id, std::slice::from_ref(&pipeline))
1885 .await
1886 .unwrap();
1887 let changed = repository.get_cached_pipelines(None).await.unwrap();
1888 assert_eq!(changed[0].metadata, pipeline.metadata);
1889 assert_eq!(changed[0].status, pipeline.status);
1890 }
1891
1892 fn jenkins_provider() -> ProviderConfig {
1893 ProviderConfig {
1894 id: None,
1895 name: "ci-jenkins".to_string(),
1896 provider_type: "jenkins".to_string(),
1897 token: String::new(),
1898 config: HashMap::new(),
1899 refresh_interval: 60,
1900 version: None,
1901 }
1902 }
1903
1904 #[tokio::test]
1905 async fn get_provider_reads_a_missing_token_entry_as_empty() {
1906 let dir = tempfile::tempdir().unwrap();
1907 let (manager, repository) = test_repository(&dir).await;
1908 let id = manager
1909 .config_backend()
1910 .create_provider(&jenkins_provider())
1911 .await
1912 .unwrap();
1913
1914 let provider = repository.get_provider(id).await.unwrap();
1915
1916 assert_eq!(provider.name, "ci-jenkins");
1917 assert_eq!(provider.token, "");
1918 }
1919
1920 struct FailingTokenStore;
1921
1922 #[async_trait::async_trait]
1923 impl TokenStore for FailingTokenStore {
1924 async fn store_token(&self, _provider_id: i64, _token: &str) -> DomainResult<()> {
1925 Ok(())
1926 }
1927
1928 async fn get_token(&self, _provider_id: i64) -> DomainResult<String> {
1929 Err(DomainError::AuthenticationFailed(
1930 "vault is locked".to_string(),
1931 ))
1932 }
1933
1934 async fn delete_token(&self, _provider_id: i64) -> DomainResult<()> {
1935 Ok(())
1936 }
1937
1938 async fn get_all_tokens(&self) -> DomainResult<HashMap<i64, String>> {
1939 Ok(HashMap::new())
1940 }
1941 }
1942
1943 #[tokio::test]
1944 async fn get_provider_fails_when_the_token_store_fails() {
1945 let dir = tempfile::tempdir().unwrap();
1946 let (manager, _) = test_repository(&dir).await;
1947 let id = manager
1948 .config_backend()
1949 .create_provider(&jenkins_provider())
1950 .await
1951 .unwrap();
1952 let repository = Repository::new(
1953 manager.config_backend(),
1954 manager.database_pool(),
1955 Arc::new(FailingTokenStore),
1956 );
1957
1958 let result = repository.get_provider(id).await;
1959
1960 assert!(matches!(result, Err(DomainError::AuthenticationFailed(_))));
1961 }
1962
1963 #[tokio::test]
1964 async fn native_run_cache_lists_the_latest_activity_first_on_every_page() {
1965 let dir = tempfile::tempdir().unwrap();
1966 let (_manager, repository) = test_repository(&dir).await;
1967 let flux = "fluxcd__1__Kustomization__apps__web";
1968 let other = "github__1__acme__web";
1969 let runs = |pipeline_id: &str| {
1970 [
1971 cached_run(pipeline_id, 100, "2026-10-22T10:00:00Z"),
1972 cached_run(pipeline_id, 200, "2026-10-20T10:00:00Z"),
1973 cached_run(pipeline_id, 300, "2026-10-21T10:00:00.500Z"),
1974 cached_run(pipeline_id, 400, "2026-10-21T10:00:00Z"),
1975 cached_run(pipeline_id, 500, "2026-10-21T10:00:00Z"),
1976 ]
1977 };
1978 repository
1979 .cache_run_history(flux, &runs(flux))
1980 .await
1981 .unwrap();
1982 repository
1983 .cache_run_history(other, &runs(other))
1984 .await
1985 .unwrap();
1986
1987 assert_eq!(
1988 run_pages(&repository, flux).await,
1989 vec![vec![100, 300], vec![500, 400], vec![200]]
1990 );
1991 assert_eq!(
1992 run_pages(&repository, other).await,
1993 vec![vec![500, 400], vec![300, 200], vec![100]]
1994 );
1995
1996 let mut reconciled_again = runs(flux)[1].clone();
1997 reconciled_again.started_at = "2026-10-23T10:00:00Z".parse().unwrap();
1998 let mut archived = runs(flux)[0].clone();
1999 archived
2000 .metadata
2001 .insert("archived".to_string(), json!(true));
2002 repository
2003 .merge_run_cache(
2004 flux,
2005 Vec::new(),
2006 vec![reconciled_again, archived],
2007 Vec::new(),
2008 )
2009 .await
2010 .unwrap();
2011
2012 assert_eq!(
2013 run_pages(&repository, flux).await,
2014 vec![vec![200, 100], vec![300, 500], vec![400]]
2015 );
2016 let newest = repository.get_cached_run_history(flux, 3).await.unwrap();
2017 assert_eq!(
2018 newest.iter().map(|run| run.run_number).collect::<Vec<_>>(),
2019 vec![200, 100, 300]
2020 );
2021 let kept = repository.get_cached_run(flux, 100).await.unwrap().unwrap();
2022 assert_eq!(kept.metadata["archived"], json!(true));
2023 assert_eq!(repository.get_cached_run_count(flux).await.unwrap(), 5);
2024 }
2025
2026 #[tokio::test]
2027 async fn sqlite_migration_backfills_the_order_key_of_cached_flux_runs() {
2028 let pool = sqlx::sqlite::SqlitePoolOptions::new()
2029 .max_connections(1)
2030 .connect("sqlite::memory:")
2031 .await
2032 .unwrap();
2033 sqlx::raw_sql(
2034 "CREATE TABLE run_history_cache (
2035 pipeline_id TEXT NOT NULL,
2036 run_number INTEGER NOT NULL,
2037 run_data TEXT NOT NULL,
2038 PRIMARY KEY (pipeline_id, run_number)
2039 )",
2040 )
2041 .execute(&pool)
2042 .await
2043 .unwrap();
2044 let timestamps = [
2045 "2026-10-21T10:00:00Z",
2046 "2026-10-21T10:00:00.1Z",
2047 "2026-10-21T10:00:00.500999Z",
2048 "2026-10-21T10:00:00.999999Z",
2049 ];
2050 let pipelines = [
2051 "fluxcd__1__Kustomization__apps__web",
2052 "github__1__acme__web",
2053 ];
2054 for pipeline_id in pipelines {
2055 for (index, timestamp) in timestamps.iter().enumerate() {
2056 sqlx::query("INSERT INTO run_history_cache (pipeline_id, run_number, run_data) VALUES (?, ?, ?)")
2057 .bind(pipeline_id)
2058 .bind((index + 1) as i64)
2059 .bind(json!({ "started_at": timestamp }).to_string())
2060 .execute(&pool).await.unwrap();
2061 }
2062 }
2063
2064 sqlx::raw_sql(include_str!(
2065 "../../../migrations/sqlite/20261021000001_run_history_started_at.sql"
2066 ))
2067 .execute(&pool)
2068 .await
2069 .unwrap();
2070
2071 let keys: Vec<(String, i64, i64)> = sqlx::query_as(
2072 "SELECT pipeline_id, run_number, started_at_ms FROM run_history_cache ORDER BY pipeline_id, run_number",
2073 ).fetch_all(&pool).await.unwrap();
2074 let expected: Vec<_> = pipelines
2075 .iter()
2076 .flat_map(|pipeline_id| {
2077 timestamps
2078 .iter()
2079 .enumerate()
2080 .map(move |(index, timestamp)| {
2081 let millis = if pipeline_id.starts_with("fluxcd__") {
2082 timestamp
2083 .parse::<DateTime<Utc>>()
2084 .unwrap()
2085 .timestamp_millis()
2086 } else {
2087 0
2088 };
2089 (pipeline_id.to_string(), (index + 1) as i64, millis)
2090 })
2091 })
2092 .collect();
2093 assert_eq!(keys, expected);
2094 }
2095}