1use std::collections::HashSet;
2use std::sync::Arc;
3
4use chrono::{
5 DateTime,
6 Utc,
7};
8
9use crate::domain::{
10 AggregatedMetrics,
11 DomainError,
12 DomainResult,
13 GlobalMetricsConfig,
14 MetricEntry,
15 MetricMetadata,
16 MetricType,
17 MetricsConfig,
18 MetricsQuery,
19 MetricsStats,
20 PipelineRun,
21 PipelineStatus,
22};
23use crate::infrastructure::database::{
24 MetricsRepository,
25 RecordMetrics,
26};
27use crate::infrastructure::deduplication::hash_pipeline_run;
28
29pub struct MetricsService {
30 repository: Arc<MetricsRepository>,
31}
32
33impl MetricsService {
34 pub fn new(repository: Arc<MetricsRepository>) -> Self {
35 Self { repository }
36 }
37
38 pub fn repository(&self) -> Arc<MetricsRepository> {
39 Arc::clone(&self.repository)
40 }
41
42 pub async fn get_global_config(&self) -> DomainResult<GlobalMetricsConfig> {
43 self.repository.get_global_config().await
44 }
45
46 pub async fn update_global_config(
47 &self, enabled: bool, default_retention_days: i64,
48 ) -> DomainResult<()> {
49 if default_retention_days < 1 {
50 return Err(DomainError::InvalidConfig(
51 "Retention days must be at least 1".to_string(),
52 ));
53 }
54
55 let previous_config = self.repository.get_global_config().await?;
56
57 self.repository
58 .update_global_config(enabled, default_retention_days)
59 .await?;
60
61 if enabled && !previous_config.enabled {
62 tracing::info!("Global metrics re-enabled - resetting all processing states");
63 self.repository.reset_all_processing_states().await?;
64 }
65
66 Ok(())
67 }
68
69 pub async fn get_pipeline_config(
70 &self, pipeline_id: &str,
71 ) -> DomainResult<Option<MetricsConfig>> {
72 self.repository.get_pipeline_config(pipeline_id).await
73 }
74
75 pub async fn get_effective_pipeline_config(
76 &self, pipeline_id: &str,
77 ) -> DomainResult<MetricsConfig> {
78 if let Some(config) = self.repository.get_pipeline_config(pipeline_id).await? {
79 Ok(config)
80 } else {
81 let global_config = self.repository.get_global_config().await?;
82 Ok(MetricsConfig {
83 pipeline_id: pipeline_id.to_string(),
84 enabled: global_config.enabled,
85 retention_days: global_config.default_retention_days,
86 created_at: Utc::now(),
87 updated_at: Utc::now(),
88 })
89 }
90 }
91
92 pub async fn update_pipeline_config(
93 &self, pipeline_id: &str, enabled: bool, retention_days: i64,
94 ) -> DomainResult<()> {
95 if retention_days < 1 {
96 return Err(DomainError::InvalidConfig(
97 "Retention days must be at least 1".to_string(),
98 ));
99 }
100
101 let previous_config = self.get_effective_pipeline_config(pipeline_id).await?;
102
103 self.repository
104 .upsert_pipeline_config(pipeline_id, enabled, retention_days)
105 .await?;
106
107 if enabled && !previous_config.enabled {
108 tracing::info!(
109 pipeline_id = %pipeline_id,
110 "Metrics re-enabled for pipeline - resetting processing state"
111 );
112 self.repository.reset_processing_state(pipeline_id).await?;
113 }
114
115 Ok(())
116 }
117
118 pub async fn extract_and_store_metrics(
119 &self, pipeline_id: &str, runs: &[PipelineRun],
120 ) -> DomainResult<usize> {
121 if runs.is_empty() {
122 return Ok(0);
123 }
124
125 let config = self.get_effective_pipeline_config(pipeline_id).await?;
126 if !config.enabled {
127 return Ok(0);
128 }
129 let retention_cutoff = chrono::Duration::try_days(config.retention_days)
130 .and_then(|retention| Utc::now().checked_sub_signed(retention))
131 .ok_or_else(|| {
132 DomainError::InvalidConfig(
133 "Metrics retention is outside the supported date range".to_string(),
134 )
135 })?;
136
137 let mut records = Vec::new();
138 let mut has_regular_runs = false;
139 for run in runs.iter().filter(|run| !is_snapshot(run)) {
140 if is_native_history(run) {
141 if is_final(&run.status) && run.started_at >= retention_cutoff {
142 records.push(record_metrics(pipeline_id, run));
143 }
144 } else {
145 has_regular_runs = true;
146 }
147 }
148 let refreshed = self
149 .repository
150 .refresh_record_metrics(pipeline_id, &records)
151 .await?;
152
153 let stored = if has_regular_runs {
154 self.store_new_runs(pipeline_id, runs, retention_cutoff)
155 .await?
156 } else {
157 0
158 };
159
160 Ok(refreshed + stored)
161 }
162
163 async fn store_new_runs(
164 &self, pipeline_id: &str, runs: &[PipelineRun], retention_cutoff: DateTime<Utc>,
165 ) -> DomainResult<usize> {
166 if runs.is_empty() {
167 return Ok(0);
168 }
169
170 let last_processed = self
171 .repository
172 .get_last_processed_run(pipeline_id)
173 .await?
174 .unwrap_or(0);
175
176 tracing::debug!(
177 pipeline_id = %pipeline_id,
178 total_runs = runs.len(),
179 last_processed = last_processed,
180 "Checking runs for metrics extraction"
181 );
182
183 let candidates: Vec<_> = runs
184 .iter()
185 .filter(|run| {
186 !is_snapshot(run)
187 && !is_native_history(run)
188 && run.run_number > last_processed
189 && run.started_at >= retention_cutoff
190 })
191 .collect();
192 let oldest_unfinished = candidates
193 .iter()
194 .filter(|run| !is_final(&run.status))
195 .map(|run| run.run_number)
196 .min();
197 let finished: Vec<_> = candidates
198 .into_iter()
199 .filter(|run| is_final(&run.status))
200 .collect();
201 let stored = if finished.is_empty() {
202 HashSet::new()
203 } else {
204 self.repository
205 .get_stored_run_numbers_after(pipeline_id, last_processed)
206 .await?
207 };
208 let mut new_runs: Vec<_> = finished
209 .iter()
210 .copied()
211 .filter(|run| !stored.contains(&run.run_number))
212 .collect();
213 new_runs.sort_by_key(|run| run.run_number);
214
215 tracing::debug!(
216 pipeline_id = %pipeline_id,
217 new_runs_count = new_runs.len(),
218 filtered_out = runs.len() - new_runs.len(),
219 "Filtered runs for processing"
220 );
221
222 if new_runs.is_empty() && !runs.is_empty() {
223 tracing::warn!(
224 pipeline_id = %pipeline_id,
225 total_runs = runs.len(),
226 last_processed_run = last_processed,
227 "No new runs to process - possible processing state corruption"
228 );
229 }
230
231 if new_runs.is_empty() {
232 return Ok(0);
233 }
234
235 const MAX_BATCH_SIZE: usize = 200;
236 let runs_to_process = if new_runs.len() > MAX_BATCH_SIZE {
237 tracing::debug!(
238 available = new_runs.len(),
239 processing = MAX_BATCH_SIZE,
240 pipeline_id = %pipeline_id,
241 "Limiting batch size"
242 );
243 &new_runs[..MAX_BATCH_SIZE]
244 } else {
245 tracing::debug!(
246 count = new_runs.len(),
247 pipeline_id = %pipeline_id,
248 last_processed = last_processed,
249 "Processing new runs"
250 );
251 &new_runs[..]
252 };
253
254 let processed_through = if runs_to_process.len() < new_runs.len() {
255 runs_to_process.last().map(|run| run.run_number)
256 } else {
257 finished.iter().map(|run| run.run_number).max()
258 }
259 .unwrap_or(last_processed);
260 let max_run_number = oldest_unfinished
261 .map_or(processed_through, |run_number| {
262 processed_through.min(run_number - 1)
263 })
264 .max(last_processed);
265
266 let mut metrics = Vec::new();
267 for run in runs_to_process {
268 metrics.extend(
269 record_metrics(pipeline_id, run)
270 .entries
271 .into_iter()
272 .flatten(),
273 );
274 }
275
276 if metrics.is_empty() {
277 return Ok(0);
278 }
279
280 let start = std::time::Instant::now();
281 let inserted = self.repository.insert_metrics_batch(metrics).await?;
282 let insert_duration = start.elapsed();
283
284 if insert_duration.as_secs() > 2 {
285 tracing::warn!(
286 duration_ms = insert_duration.as_millis(),
287 count = inserted,
288 "Slow metrics insert"
289 );
290 }
291
292 self.repository
293 .update_last_processed_run(pipeline_id, max_run_number)
294 .await?;
295
296 tracing::debug!(
297 inserted = inserted,
298 pipeline_id = %pipeline_id,
299 duration_ms = insert_duration.as_millis(),
300 last_run = max_run_number,
301 "Stored metrics"
302 );
303
304 Ok(inserted)
305 }
306
307 pub async fn query_metrics(&self, query: MetricsQuery) -> DomainResult<Vec<MetricEntry>> {
308 self.repository.query_metrics(query).await
309 }
310
311 pub async fn query_aggregated_metrics(
312 &self, query: MetricsQuery,
313 ) -> DomainResult<AggregatedMetrics> {
314 self.repository.query_aggregated_metrics(query).await
315 }
316
317 pub async fn cleanup_old_metrics(&self) -> DomainResult<usize> {
318 self.repository.delete_old_metrics(None).await
319 }
320
321 pub async fn get_storage_stats(&self) -> DomainResult<MetricsStats> {
322 self.repository.get_storage_stats().await
323 }
324
325 pub async fn flush_metrics(
326 &self, pipeline_id: Option<&str>, skip_vacuum: bool,
327 ) -> DomainResult<usize> {
328 let result = self
329 .repository
330 .flush_metrics(pipeline_id, skip_vacuum)
331 .await?;
332
333 if let Some(pipeline) = pipeline_id {
334 self.repository.reset_processing_state(pipeline).await?;
335 } else {
336 self.repository.reset_all_processing_states().await?;
337 }
338
339 Ok(result)
340 }
341
342 pub async fn reset_pipeline_processing(&self, pipeline_id: &str) -> DomainResult<()> {
343 tracing::info!(pipeline_id = %pipeline_id, "Resetting metrics processing state");
344 self.repository.reset_processing_state(pipeline_id).await?;
345 Ok(())
346 }
347}
348
349fn is_snapshot(run: &PipelineRun) -> bool {
350 run.metadata
351 .get("snapshot")
352 .and_then(serde_json::Value::as_bool)
353 == Some(true)
354}
355
356fn metadata_text<'a>(run: &'a PipelineRun, key: &str) -> Option<&'a str> {
357 run.metadata.get(key).and_then(serde_json::Value::as_str)
358}
359
360fn is_native_history(run: &PipelineRun) -> bool {
361 metadata_text(run, "record_source") == Some("status.history")
362 && metadata_text(run, "history_scope") == Some("bounded_reconciliation")
363}
364
365fn is_final(status: &PipelineStatus) -> bool {
366 matches!(
367 status,
368 PipelineStatus::Success
369 | PipelineStatus::Failed
370 | PipelineStatus::Cancelled
371 | PipelineStatus::Skipped
372 )
373}
374
375fn record_metrics(pipeline_id: &str, run: &PipelineRun) -> RecordMetrics {
376 let status_str = run.status.as_str();
377
378 let run_hash = hash_pipeline_run(
379 run.run_number,
380 status_str,
381 run.branch.as_deref(),
382 &run.started_at.to_rfc3339(),
383 run.duration_seconds,
384 run.commit_sha.as_deref(),
385 );
386
387 let metadata = MetricMetadata {
388 status: Some(status_str.to_string()),
389 branch: run.branch.clone(),
390 repository: None,
391 actor: run.actor.clone(),
392 }
393 .to_json();
394
395 let entry = |metric_type, value| MetricEntry {
396 id: 0,
397 pipeline_id: pipeline_id.to_string(),
398 run_number: run.run_number,
399 timestamp: run.started_at,
400 metric_type,
401 value,
402 metadata: Some(metadata.clone()),
403 created_at: Utc::now(),
404 run_hash: Some(run_hash.clone()),
405 };
406
407 let success_value = match run.status {
408 PipelineStatus::Success => Some(100.0),
409 PipelineStatus::Failed | PipelineStatus::Cancelled => Some(0.0),
410 _ => None,
411 };
412
413 let entries = [
414 run.duration_seconds
415 .map(|duration| entry(MetricType::RunDuration, duration as f64)),
416 success_value.map(|value| entry(MetricType::SuccessRate, value)),
417 success_value.map(|_| entry(MetricType::RunFrequency, 1.0)),
418 ];
419
420 RecordMetrics {
421 run_number: run.run_number,
422 run_hash,
423 entries,
424 }
425}
426
427#[cfg(test)]
428mod tests {
429 use serde_json::json;
430
431 use super::*;
432
433 #[tokio::test]
434 async fn state_snapshots_do_not_count_as_failed_reconciliations() {
435 let dir = tempfile::tempdir().unwrap();
436 let pool = crate::infrastructure::database::init_database(dir.path().join("metrics.db"))
437 .await
438 .unwrap();
439 let service = MetricsService::new(Arc::new(MetricsRepository::new(pool.clone())));
440 service.update_global_config(true, 30).await.unwrap();
441 let mut run: PipelineRun = serde_json::from_value(json!({
442 "id": "snapshot", "pipeline_id": "fluxcd__1__Kustomization__apps__web", "run_number": 1,
443 "status": "failed", "started_at": Utc::now().to_rfc3339(), "logs_url": "",
444 "metadata": { "snapshot": true }
445 }))
446 .unwrap();
447
448 service
449 .extract_and_store_metrics(&run.pipeline_id, std::slice::from_ref(&run))
450 .await
451 .unwrap();
452 let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pipeline_metrics")
453 .fetch_one(&pool)
454 .await
455 .unwrap();
456 assert_eq!(count, 0);
457
458 run.metadata.clear();
459 run.status = PipelineStatus::Success;
460 service
461 .extract_and_store_metrics(&run.pipeline_id, std::slice::from_ref(&run))
462 .await
463 .unwrap();
464 let success: f64 = sqlx::query_scalar(
465 "SELECT value FROM pipeline_metrics WHERE metric_type = 'success_rate'",
466 )
467 .fetch_one(&pool)
468 .await
469 .unwrap();
470 assert_eq!(success, 100.0);
471 }
472
473 #[tokio::test]
474 async fn runs_past_retention_are_not_recorded() {
475 let dir = tempfile::tempdir().unwrap();
476 let pool = crate::infrastructure::database::init_database(dir.path().join("metrics.db"))
477 .await
478 .unwrap();
479 let service = MetricsService::new(Arc::new(MetricsRepository::new(pool)));
480 service.update_global_config(true, 7).await.unwrap();
481 let pipeline_id = "github__1__o__r__1";
482 let run = |run_number: i64, days_ago: i64| -> PipelineRun {
483 serde_json::from_value(json!({
484 "id": format!("{pipeline_id}__{run_number}"), "pipeline_id": pipeline_id,
485 "run_number": run_number, "status": "success",
486 "started_at": (Utc::now() - chrono::Duration::days(days_ago)).to_rfc3339(),
487 "logs_url": ""
488 }))
489 .unwrap()
490 };
491
492 service
493 .extract_and_store_metrics(pipeline_id, &[run(2, 1), run(1, 10)])
494 .await
495 .unwrap();
496
497 let runs: Vec<_> = stored(&service, pipeline_id)
498 .await
499 .iter()
500 .map(|entry| entry.run_number)
501 .collect();
502 assert_eq!(runs, [2, 2]);
503 }
504
505 #[tokio::test]
506 async fn run_that_finishes_after_a_newer_run_gets_metrics() {
507 let dir = tempfile::tempdir().unwrap();
508 let service = sqlite_service(&dir).await;
509 service.update_global_config(true, 30).await.unwrap();
510 let pipeline_id = "github__1__acme__app__deploy";
511 let run = |run_number: i64, status: &str| -> PipelineRun {
512 serde_json::from_value(json!({
513 "id": format!("{pipeline_id}__{run_number}"), "pipeline_id": pipeline_id,
514 "run_number": run_number, "status": status,
515 "started_at": Utc::now().to_rfc3339(), "logs_url": "",
516 "duration_seconds": 60
517 }))
518 .unwrap()
519 };
520
521 let first = service
522 .extract_and_store_metrics(pipeline_id, &[run(11, "success"), run(10, "running")])
523 .await
524 .unwrap();
525 assert_eq!(first, 3);
526 let mut rerun_11 = run(11, "success");
527 rerun_11.duration_seconds = Some(90);
528 let second = service
529 .extract_and_store_metrics(pipeline_id, &[rerun_11, run(10, "failed")])
530 .await
531 .unwrap();
532 assert_eq!(second, 3);
533 let repeated = service
534 .extract_and_store_metrics(pipeline_id, &[run(11, "success"), run(10, "failed")])
535 .await
536 .unwrap();
537 assert_eq!(repeated, 0);
538
539 let entries = stored(&service, pipeline_id).await;
540 assert_eq!(entries.len(), 6);
541 assert_eq!(
542 find(&entries, 10, MetricType::SuccessRate).unwrap().value,
543 0.0
544 );
545 assert_eq!(
546 find(&entries, 11, MetricType::SuccessRate).unwrap().value,
547 100.0
548 );
549 }
550
551 fn native_run(
552 pipeline_id: &str, run_number: i64, started_at: &str, duration_seconds: Option<i64>,
553 ) -> PipelineRun {
554 let mut run: PipelineRun = serde_json::from_value(json!({
555 "id": format!("{pipeline_id}__{run_number}"), "pipeline_id": pipeline_id,
556 "run_number": run_number, "status": "success", "started_at": started_at,
557 "logs_url": "",
558 "metadata": {
559 "record_source": "status.history", "history_scope": "bounded_reconciliation",
560 "total_reconciliations": 1
561 }
562 }))
563 .unwrap();
564 run.duration_seconds = duration_seconds;
565 run
566 }
567
568 fn snapshot_run(pipeline_id: &str, run_number: i64) -> PipelineRun {
569 serde_json::from_value(json!({
570 "id": format!("{pipeline_id}__{run_number}"), "pipeline_id": pipeline_id,
571 "run_number": run_number, "status": "failed", "started_at": "2026-10-21T00:30:00Z",
572 "logs_url": "",
573 "metadata": {
574 "record_source": "current_snapshot", "history_scope": "bounded_reconciliation",
575 "snapshot": true
576 }
577 }))
578 .unwrap()
579 }
580
581 async fn stored(service: &MetricsService, pipeline_id: &str) -> Vec<MetricEntry> {
582 service
583 .query_metrics(MetricsQuery {
584 pipeline_id: Some(pipeline_id.to_string()),
585 ..Default::default()
586 })
587 .await
588 .unwrap()
589 }
590
591 fn find(
592 entries: &[MetricEntry], run_number: i64, metric_type: MetricType,
593 ) -> Option<&MetricEntry> {
594 entries
595 .iter()
596 .find(|entry| entry.run_number == run_number && entry.metric_type == metric_type)
597 }
598
599 fn ids(entries: &[MetricEntry], run_number: i64) -> Vec<i64> {
600 let mut ids: Vec<i64> = entries
601 .iter()
602 .filter(|entry| entry.run_number == run_number)
603 .map(|entry| entry.id)
604 .collect();
605 ids.sort_unstable();
606 ids
607 }
608
609 async fn assert_native_history_refresh(service: &MetricsService, pipeline_id: &str) {
610 service
611 .update_pipeline_config(pipeline_id, true, 30)
612 .await
613 .unwrap();
614 let (older, newer) = (1_760_000_000_000_011, 1_760_000_600_000_022);
615 let started = chrono::DateTime::from_timestamp_millis(Utc::now().timestamp_millis())
616 .unwrap()
617 - chrono::Duration::hours(1);
618 let snapshot = snapshot_run(pipeline_id, newer + 1);
619 let newest = native_run(
620 pipeline_id,
621 newer,
622 &(started + chrono::Duration::minutes(10)).to_rfc3339(),
623 Some(7),
624 );
625 let mut grouped = native_run(pipeline_id, older, &started.to_rfc3339(), Some(10));
626 let batch = |grouped: &PipelineRun| vec![snapshot.clone(), newest.clone(), grouped.clone()];
627
628 let first = service
629 .extract_and_store_metrics(pipeline_id, &batch(&grouped))
630 .await
631 .unwrap();
632 assert_eq!(first, 6);
633 let before = stored(service, pipeline_id).await;
634 assert_eq!(before.len(), 6);
635 assert!(before.iter().all(|entry| entry.run_number != newer + 1));
636
637 let repeated = service
638 .extract_and_store_metrics(pipeline_id, &batch(&grouped))
639 .await
640 .unwrap();
641 assert_eq!(repeated, 0);
642 let unchanged = stored(service, pipeline_id).await;
643 assert_eq!(ids(&unchanged, older), ids(&before, older));
644 assert_eq!(ids(&unchanged, newer), ids(&before, newer));
645
646 grouped.started_at = started + chrono::Duration::minutes(20);
647 grouped.duration_seconds = Some(25);
648 grouped
649 .metadata
650 .insert("total_reconciliations".to_string(), json!(9));
651 service
652 .extract_and_store_metrics(pipeline_id, &batch(&grouped))
653 .await
654 .unwrap();
655 let refreshed = stored(service, pipeline_id).await;
656 assert_eq!(refreshed.len(), 6);
657 let duration = find(&refreshed, older, MetricType::RunDuration).unwrap();
658 assert_eq!(duration.value, 25.0);
659 assert_eq!(duration.timestamp, grouped.started_at);
660 assert_eq!(ids(&refreshed, newer), ids(&before, newer));
661 let frequency: f64 = refreshed
662 .iter()
663 .filter(|entry| entry.metric_type == MetricType::RunFrequency)
664 .map(|entry| entry.value)
665 .sum();
666 assert_eq!(frequency, 2.0);
667
668 grouped.started_at = started + chrono::Duration::minutes(21);
669 grouped.duration_seconds = None;
670 service
671 .extract_and_store_metrics(pipeline_id, &batch(&grouped))
672 .await
673 .unwrap();
674 let without_duration = stored(service, pipeline_id).await;
675 assert_eq!(without_duration.len(), 5);
676 assert!(find(&without_duration, older, MetricType::RunDuration).is_none());
677 assert_eq!(
678 find(&without_duration, older, MetricType::SuccessRate)
679 .unwrap()
680 .timestamp,
681 grouped.started_at
682 );
683
684 let settled = service
685 .extract_and_store_metrics(pipeline_id, &batch(&grouped))
686 .await
687 .unwrap();
688 assert_eq!(settled, 0);
689 assert_eq!(stored(service, pipeline_id).await.len(), 5);
690
691 grouped.status = PipelineStatus::Skipped;
692 let removed = service
693 .extract_and_store_metrics(pipeline_id, &batch(&grouped))
694 .await
695 .unwrap();
696 assert_eq!(removed, 2);
697 assert!(stored(service, pipeline_id)
698 .await
699 .iter()
700 .all(|entry| entry.run_number != older));
701 }
702
703 async fn assert_concurrent_refresh_is_deduplicated(
704 service: &MetricsService, pipeline_id: &str,
705 ) {
706 service
707 .update_pipeline_config(pipeline_id, true, 30)
708 .await
709 .unwrap();
710 let runs = [native_run(
711 pipeline_id,
712 1_760_000_000_000_033,
713 &Utc::now().to_rfc3339(),
714 Some(5),
715 )];
716
717 let inserted: usize = futures::future::join_all(
718 (0..8).map(|_| service.extract_and_store_metrics(pipeline_id, &runs)),
719 )
720 .await
721 .into_iter()
722 .map(Result::unwrap)
723 .sum();
724
725 assert_eq!(inserted, 3);
726 assert_eq!(stored(service, pipeline_id).await.len(), 3);
727 }
728
729 async fn sqlite_service(dir: &tempfile::TempDir) -> MetricsService {
730 let pool = crate::infrastructure::database::init_database(dir.path().join("metrics.db"))
731 .await
732 .unwrap();
733 MetricsService::new(Arc::new(MetricsRepository::new(pool)))
734 }
735
736 #[tokio::test]
737 async fn native_history_refreshes_changed_records_without_inflating_counts() {
738 let dir = tempfile::tempdir().unwrap();
739 let service = sqlite_service(&dir).await;
740
741 assert_native_history_refresh(&service, "fluxcd__1__Kustomization__apps__web").await;
742 }
743
744 #[tokio::test]
745 async fn concurrent_native_refreshes_store_each_metric_once() {
746 let dir = tempfile::tempdir().unwrap();
747 let service = sqlite_service(&dir).await;
748
749 assert_concurrent_refresh_is_deduplicated(&service, "fluxcd__1__Kustomization__apps__api")
750 .await;
751 }
752
753 #[tokio::test]
754 async fn runs_without_native_provenance_keep_their_first_observation() {
755 let dir = tempfile::tempdir().unwrap();
756 let service = sqlite_service(&dir).await;
757 let pipeline_id = "github__1__acme__repo";
758 service
759 .update_pipeline_config(pipeline_id, true, 30)
760 .await
761 .unwrap();
762 let mut run = native_run(pipeline_id, 5, "2026-10-21T00:00:00Z", Some(10));
763 run.metadata.clear();
764
765 let first = service
766 .extract_and_store_metrics(pipeline_id, std::slice::from_ref(&run))
767 .await
768 .unwrap();
769 assert_eq!(first, 3);
770
771 run.started_at = "2026-10-21T00:20:00Z".parse().unwrap();
772 run.duration_seconds = Some(99);
773 let second = service
774 .extract_and_store_metrics(pipeline_id, std::slice::from_ref(&run))
775 .await
776 .unwrap();
777 assert_eq!(second, 0);
778
779 let entries = stored(&service, pipeline_id).await;
780 assert_eq!(entries.len(), 3);
781 assert_eq!(
782 find(&entries, 5, MetricType::RunDuration).unwrap().value,
783 10.0
784 );
785 }
786
787 #[cfg(feature = "postgres")]
788 #[tokio::test]
789 #[ignore = "requires TEST_DATABASE_URL"]
790 async fn postgres_native_history_refreshes_changed_records_atomically() {
791 use crate::infrastructure::database::DatabasePool;
792
793 let url = std::env::var("TEST_DATABASE_URL")
794 .expect("TEST_DATABASE_URL must be set to run the PostgreSQL tests");
795 let pool = crate::infrastructure::database::init_postgres_database(&url)
796 .await
797 .unwrap();
798 let service = MetricsService::new(Arc::new(MetricsRepository::new_from_pool(
799 DatabasePool::Postgres(pool.clone()),
800 )));
801 let suffix = Utc::now().timestamp_nanos_opt().unwrap();
802 let refresh_id = format!("fluxcd__postgres__Kustomization__refresh__{suffix}");
803 let concurrent_id = format!("fluxcd__postgres__Kustomization__concurrent__{suffix}");
804 let retention_id = format!("fluxcd__postgres__HelmRelease__retention__{suffix}");
805
806 assert_native_history_refresh(&service, &refresh_id).await;
807 assert_concurrent_refresh_is_deduplicated(&service, &concurrent_id).await;
808 assert_native_retention(&service, &retention_id).await;
809
810 for pipeline_id in [&refresh_id, &concurrent_id, &retention_id] {
811 service
812 .flush_metrics(Some(pipeline_id.as_str()), true)
813 .await
814 .unwrap();
815 sqlx::query("DELETE FROM metrics_config WHERE pipeline_id = $1")
816 .bind(pipeline_id)
817 .execute(&pool)
818 .await
819 .unwrap();
820 }
821 pool.close().await;
822 }
823
824 async fn assert_native_retention(service: &MetricsService, pipeline_id: &str) {
825 let now = chrono::DateTime::from_timestamp_millis(Utc::now().timestamp_millis()).unwrap();
826 service
827 .update_pipeline_config(pipeline_id, true, 30)
828 .await
829 .unwrap();
830 let mut expired = native_run(
831 pipeline_id,
832 1,
833 &(now - chrono::Duration::days(2)).to_rfc3339(),
834 Some(10),
835 );
836 let recent = native_run(
837 pipeline_id,
838 2,
839 &(now - chrono::Duration::hours(1)).to_rfc3339(),
840 Some(5),
841 );
842 service
843 .extract_and_store_metrics(pipeline_id, &[expired.clone(), recent.clone()])
844 .await
845 .unwrap();
846 assert_eq!(stored(service, pipeline_id).await.len(), 6);
847
848 service
849 .update_pipeline_config(pipeline_id, true, 1)
850 .await
851 .unwrap();
852 assert_eq!(
853 service
854 .repository
855 .delete_old_metrics(Some(pipeline_id))
856 .await
857 .unwrap(),
858 3
859 );
860 let remaining = stored(service, pipeline_id).await;
861 assert_eq!(remaining.len(), 3);
862 assert!(remaining
863 .iter()
864 .all(|entry| entry.run_number == recent.run_number));
865
866 let restored = service
867 .extract_and_store_metrics(pipeline_id, &[expired.clone(), recent.clone()])
868 .await
869 .unwrap();
870 assert_eq!(restored, 0);
871 let after_read = stored(service, pipeline_id).await;
872 assert_eq!(after_read.len(), 3);
873 assert_eq!(
874 ids(&after_read, recent.run_number),
875 ids(&remaining, recent.run_number)
876 );
877
878 expired.started_at = now - chrono::Duration::minutes(1);
879 expired.duration_seconds = Some(25);
880 service
881 .extract_and_store_metrics(pipeline_id, &[expired.clone(), recent])
882 .await
883 .unwrap();
884 let renewed = stored(service, pipeline_id).await;
885 assert_eq!(renewed.len(), 6);
886 let metric = find(&renewed, expired.run_number, MetricType::RunDuration).unwrap();
887 assert_eq!(metric.value, 25.0);
888 assert_eq!(metric.timestamp, expired.started_at);
889 }
890
891 #[tokio::test]
892 async fn expired_native_history_does_not_restore_deleted_metrics() {
893 let dir = tempfile::tempdir().unwrap();
894 let service = sqlite_service(&dir).await;
895 assert_native_retention(&service, "fluxcd__1__HelmRelease__apps__retention").await;
896 }
897}