Skip to main content

pipedash_core/infrastructure/database/
repository.rs

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