1use std::collections::HashSet;
2
3use chrono::{
4 DateTime,
5 Duration,
6 Utc,
7};
8use sqlx::Row as SqlxRow;
9
10use super::DatabasePool;
11use crate::domain::{
12 AggregatedMetric,
13 AggregatedMetrics,
14 AggregationPeriod,
15 AggregationType,
16 DomainError,
17 DomainResult,
18 GlobalMetricsConfig,
19 MetricEntry,
20 MetricType,
21 MetricsConfig,
22 MetricsConfigExport,
23 MetricsQuery,
24 MetricsStats,
25 PipelineMetricsStats,
26};
27
28pub struct RecordMetrics {
29 pub run_number: i64,
30 pub run_hash: String,
31 pub entries: [Option<MetricEntry>; 3],
32}
33
34const METRICS_LOCK_NAMESPACE: i32 = 0x6d65_7472;
35
36fn metadata_json(metric: &MetricEntry) -> String {
37 metric
38 .metadata
39 .as_ref()
40 .map_or_else(|| "null".to_string(), ToString::to_string)
41}
42
43pub struct MetricsRepository {
44 pool: DatabasePool,
45}
46
47impl MetricsRepository {
48 pub fn new(pool: sqlx::SqlitePool) -> Self {
49 Self {
50 pool: DatabasePool::Sqlite(pool),
51 }
52 }
53
54 pub fn new_from_pool(pool: DatabasePool) -> Self {
55 Self { pool }
56 }
57
58 fn placeholder(&self, index: usize) -> String {
59 match &self.pool {
60 DatabasePool::Sqlite(_) => "?".to_string(),
61 DatabasePool::Postgres(_) => format!("${}", index),
62 }
63 }
64
65 fn datetime_now(&self) -> &'static str {
66 match &self.pool {
67 DatabasePool::Sqlite(_) => "datetime('now')",
68 DatabasePool::Postgres(_) => "NOW()",
69 }
70 }
71
72 pub async fn list_pipeline_ids(&self) -> DomainResult<Vec<String>> {
73 let sql = "SELECT pipeline_id FROM pipeline_metrics
74 UNION SELECT pipeline_id FROM metrics_processing_state";
75 match &self.pool {
76 DatabasePool::Sqlite(p) => sqlx::query_scalar::<_, String>(sql).fetch_all(p).await,
77 DatabasePool::Postgres(p) => sqlx::query_scalar::<_, String>(sql).fetch_all(p).await,
78 }
79 .map_err(|e| DomainError::DatabaseError(e.to_string()))
80 }
81
82 pub async fn get_global_config(&self) -> DomainResult<GlobalMetricsConfig> {
83 let result = match &self.pool {
84 DatabasePool::Sqlite(p) => {
85 sqlx::query(
86 "SELECT enabled, default_retention_days, updated_at FROM metrics_global_config WHERE id = 1",
87 )
88 .fetch_optional(p)
89 .await
90 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
91 .map(|row| self.parse_global_config_row_sqlite(&row))
92 .transpose()?
93 }
94 DatabasePool::Postgres(p) => {
95 sqlx::query(
96 "SELECT enabled, default_retention_days::BIGINT, updated_at FROM metrics_global_config WHERE id = 1",
97 )
98 .fetch_optional(p)
99 .await
100 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
101 .map(|row| self.parse_global_config_row_postgres(&row))
102 .transpose()?
103 }
104 };
105
106 Ok(result.unwrap_or_default())
107 }
108
109 fn parse_global_config_row_sqlite(
110 &self, row: &sqlx::sqlite::SqliteRow,
111 ) -> DomainResult<GlobalMetricsConfig> {
112 let enabled: i64 = row
113 .try_get(0)
114 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
115 let default_retention_days: i64 = row
116 .try_get(1)
117 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
118 let updated_at_str: String = row
119 .try_get(2)
120 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
121
122 Ok(GlobalMetricsConfig {
123 enabled: enabled != 0,
124 default_retention_days,
125 updated_at: DateTime::parse_from_rfc3339(&updated_at_str)
126 .map(|dt| dt.with_timezone(&Utc))
127 .unwrap_or_else(|_| Utc::now()),
128 })
129 }
130
131 fn parse_global_config_row_postgres(
132 &self, row: &sqlx::postgres::PgRow,
133 ) -> DomainResult<GlobalMetricsConfig> {
134 let enabled: bool = row
135 .try_get(0)
136 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
137 let default_retention_days: i64 = row
138 .try_get(1)
139 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
140 let updated_at: DateTime<Utc> = row
141 .try_get(2)
142 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
143
144 Ok(GlobalMetricsConfig {
145 enabled,
146 default_retention_days,
147 updated_at,
148 })
149 }
150
151 pub async fn update_global_config(
152 &self, enabled: bool, default_retention_days: i64,
153 ) -> DomainResult<()> {
154 let sql = format!(
155 "UPDATE metrics_global_config SET enabled = {}, default_retention_days = {}, updated_at = {} WHERE id = 1",
156 self.placeholder(1),
157 self.placeholder(2),
158 self.datetime_now()
159 );
160
161 match &self.pool {
162 DatabasePool::Sqlite(p) => {
163 sqlx::query(sqlx::AssertSqlSafe(sql))
164 .bind(enabled as i64)
165 .bind(default_retention_days)
166 .execute(p)
167 .await
168 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
169 }
170 DatabasePool::Postgres(p) => {
171 sqlx::query(sqlx::AssertSqlSafe(sql))
172 .bind(enabled)
173 .bind(default_retention_days)
174 .execute(p)
175 .await
176 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
177 }
178 }
179
180 Ok(())
181 }
182
183 pub async fn get_pipeline_config(
184 &self, pipeline_id: &str,
185 ) -> DomainResult<Option<MetricsConfig>> {
186 match &self.pool {
187 DatabasePool::Sqlite(p) => {
188 let result = sqlx::query(
189 "SELECT pipeline_id, enabled, retention_days, created_at, updated_at FROM metrics_config WHERE pipeline_id = ?",
190 )
191 .bind(pipeline_id)
192 .fetch_optional(p)
193 .await
194 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
195
196 match result {
197 Some(row) => Ok(Some(self.metrics_config_from_sqlite_row(&row)?)),
198 None => Ok(None),
199 }
200 }
201 DatabasePool::Postgres(p) => {
202 let result = sqlx::query(
203 "SELECT pipeline_id, enabled, retention_days::BIGINT, created_at, updated_at FROM metrics_config WHERE pipeline_id = $1",
204 )
205 .bind(pipeline_id)
206 .fetch_optional(p)
207 .await
208 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
209
210 match result {
211 Some(row) => Ok(Some(self.metrics_config_from_postgres_row(&row)?)),
212 None => Ok(None),
213 }
214 }
215 }
216 }
217
218 pub async fn upsert_pipeline_config(
219 &self, pipeline_id: &str, enabled: bool, retention_days: i64,
220 ) -> DomainResult<()> {
221 match &self.pool {
222 DatabasePool::Sqlite(p) => {
223 let now = Utc::now().to_rfc3339();
224 sqlx::query(
225 "INSERT INTO metrics_config (pipeline_id, enabled, retention_days, created_at, updated_at)
226 VALUES (?, ?, ?, ?, ?)
227 ON CONFLICT(pipeline_id) DO UPDATE SET
228 enabled = excluded.enabled,
229 retention_days = excluded.retention_days,
230 updated_at = excluded.updated_at",
231 )
232 .bind(pipeline_id)
233 .bind(enabled as i64)
234 .bind(retention_days)
235 .bind(&now)
236 .bind(&now)
237 .execute(p)
238 .await
239 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
240 }
241 DatabasePool::Postgres(p) => {
242 let now = Utc::now();
243 sqlx::query(
244 "INSERT INTO metrics_config (pipeline_id, enabled, retention_days, created_at, updated_at)
245 VALUES ($1, $2, $3, $4, $5)
246 ON CONFLICT(pipeline_id) DO UPDATE SET
247 enabled = excluded.enabled,
248 retention_days = excluded.retention_days,
249 updated_at = excluded.updated_at",
250 )
251 .bind(pipeline_id)
252 .bind(enabled)
253 .bind(retention_days)
254 .bind(now)
255 .bind(now)
256 .execute(p)
257 .await
258 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
259 }
260 }
261
262 Ok(())
263 }
264
265 pub async fn get_last_processed_run(&self, pipeline_id: &str) -> DomainResult<Option<i64>> {
266 let result = match &self.pool {
267 DatabasePool::Sqlite(p) => {
268 sqlx::query_scalar::<_, i64>(
269 "SELECT last_processed_run_number FROM metrics_processing_state WHERE pipeline_id = ?",
270 )
271 .bind(pipeline_id)
272 .fetch_optional(p)
273 .await
274 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
275 }
276 DatabasePool::Postgres(p) => {
277 sqlx::query_scalar::<_, i64>(
278 "SELECT last_processed_run_number::BIGINT FROM metrics_processing_state WHERE pipeline_id = $1",
279 )
280 .bind(pipeline_id)
281 .fetch_optional(p)
282 .await
283 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
284 }
285 };
286
287 Ok(result)
288 }
289
290 pub async fn get_stored_run_numbers_after(
291 &self, pipeline_id: &str, run_number: i64,
292 ) -> DomainResult<HashSet<i64>> {
293 let result = match &self.pool {
294 DatabasePool::Sqlite(p) => {
295 sqlx::query_scalar::<_, i64>(
296 "SELECT DISTINCT run_number FROM pipeline_metrics WHERE pipeline_id = ? AND run_number > ?",
297 )
298 .bind(pipeline_id)
299 .bind(run_number)
300 .fetch_all(p)
301 .await
302 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
303 }
304 DatabasePool::Postgres(p) => {
305 sqlx::query_scalar::<_, i64>(
306 "SELECT DISTINCT run_number::BIGINT FROM pipeline_metrics WHERE pipeline_id = $1 AND run_number > $2",
307 )
308 .bind(pipeline_id)
309 .bind(run_number)
310 .fetch_all(p)
311 .await
312 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
313 }
314 };
315
316 Ok(result.into_iter().collect())
317 }
318
319 pub async fn update_last_processed_run(
320 &self, pipeline_id: &str, run_number: i64,
321 ) -> DomainResult<()> {
322 match &self.pool {
323 DatabasePool::Sqlite(p) => {
324 let now = Utc::now().to_rfc3339();
325 sqlx::query(
326 "INSERT INTO metrics_processing_state (pipeline_id, last_processed_run_number, last_processed_at, updated_at)
327 VALUES (?, ?, ?, ?)
328 ON CONFLICT(pipeline_id) DO UPDATE SET
329 last_processed_run_number = excluded.last_processed_run_number,
330 last_processed_at = excluded.last_processed_at,
331 updated_at = excluded.updated_at"
332 )
333 .bind(pipeline_id)
334 .bind(run_number)
335 .bind(&now)
336 .bind(&now)
337 .execute(p)
338 .await
339 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
340 }
341 DatabasePool::Postgres(p) => {
342 let now = Utc::now();
343 sqlx::query(
344 "INSERT INTO metrics_processing_state (pipeline_id, last_processed_run_number, last_processed_at, updated_at)
345 VALUES ($1, $2, $3, $4)
346 ON CONFLICT(pipeline_id) DO UPDATE SET
347 last_processed_run_number = excluded.last_processed_run_number,
348 last_processed_at = excluded.last_processed_at,
349 updated_at = excluded.updated_at"
350 )
351 .bind(pipeline_id)
352 .bind(run_number)
353 .bind(now)
354 .bind(now)
355 .execute(p)
356 .await
357 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
358 }
359 }
360
361 Ok(())
362 }
363
364 pub async fn reset_processing_state(&self, pipeline_id: &str) -> DomainResult<()> {
365 let sql = format!(
366 "DELETE FROM metrics_processing_state WHERE pipeline_id = {}",
367 self.placeholder(1)
368 );
369
370 match &self.pool {
371 DatabasePool::Sqlite(p) => {
372 sqlx::query(sqlx::AssertSqlSafe(sql))
373 .bind(pipeline_id)
374 .execute(p)
375 .await
376 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
377 }
378 DatabasePool::Postgres(p) => {
379 sqlx::query(sqlx::AssertSqlSafe(sql))
380 .bind(pipeline_id)
381 .execute(p)
382 .await
383 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
384 }
385 }
386
387 tracing::debug!(pipeline_id = %pipeline_id, "Reset processing state for pipeline");
388 Ok(())
389 }
390
391 pub async fn reset_all_processing_states(&self) -> DomainResult<()> {
392 match &self.pool {
393 DatabasePool::Sqlite(p) => {
394 sqlx::query("DELETE FROM metrics_processing_state")
395 .execute(p)
396 .await
397 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
398 }
399 DatabasePool::Postgres(p) => {
400 sqlx::query("DELETE FROM metrics_processing_state")
401 .execute(p)
402 .await
403 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
404 }
405 }
406
407 tracing::debug!("Reset all processing states");
408 Ok(())
409 }
410
411 pub async fn insert_metrics_batch(&self, metrics: Vec<MetricEntry>) -> DomainResult<usize> {
412 if metrics.is_empty() {
413 return Ok(0);
414 }
415
416 let start = std::time::Instant::now();
417 let metrics_count = metrics.len();
418 let mut total_inserted = 0;
419
420 const BATCH_SIZE: usize = 100;
421
422 match &self.pool {
423 DatabasePool::Sqlite(p) => {
424 let mut tx = p
425 .begin()
426 .await
427 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
428
429 for chunk in metrics.chunks(BATCH_SIZE) {
430 let prepared_data: Vec<_> = chunk
431 .iter()
432 .map(|metric| {
433 let metadata_json = metric
434 .metadata
435 .as_ref()
436 .map(|m| m.to_string())
437 .unwrap_or_else(|| "null".to_string());
438
439 (
440 metric.pipeline_id.clone(),
441 metric.run_number,
442 metric.timestamp.to_rfc3339(),
443 metric.metric_type.as_str().to_string(),
444 metric.value,
445 metadata_json,
446 metric.created_at.to_rfc3339(),
447 metric.run_hash.clone(),
448 )
449 })
450 .collect();
451
452 let values_clause = prepared_data
453 .iter()
454 .map(|_| "(?, ?, ?, ?, ?, ?, ?, ?)")
455 .collect::<Vec<_>>()
456 .join(", ");
457
458 let sql = format!(
459 "INSERT OR IGNORE INTO pipeline_metrics (pipeline_id, run_number, timestamp, metric_type, value, metadata_json, created_at, run_hash)
460 VALUES {}",
461 values_clause
462 );
463
464 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
465 for (
466 pipeline_id,
467 run_number,
468 timestamp,
469 metric_type,
470 value,
471 metadata_json,
472 created_at,
473 run_hash,
474 ) in &prepared_data
475 {
476 query = query
477 .bind(pipeline_id)
478 .bind(run_number)
479 .bind(timestamp)
480 .bind(metric_type)
481 .bind(value)
482 .bind(metadata_json)
483 .bind(created_at)
484 .bind(run_hash);
485 }
486
487 let result = query
488 .execute(&mut *tx)
489 .await
490 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
491
492 total_inserted += result.rows_affected() as usize;
493 }
494
495 tx.commit()
496 .await
497 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
498 }
499 DatabasePool::Postgres(p) => {
500 let mut tx = p
501 .begin()
502 .await
503 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
504
505 for chunk in metrics.chunks(BATCH_SIZE) {
506 let prepared_data: Vec<_> = chunk
507 .iter()
508 .map(|metric| {
509 let metadata_json = metric
510 .metadata
511 .as_ref()
512 .map(|m| m.to_string())
513 .unwrap_or_else(|| "null".to_string());
514
515 (
516 metric.pipeline_id.clone(),
517 metric.run_number,
518 metric.timestamp,
519 metric.metric_type.as_str().to_string(),
520 metric.value,
521 metadata_json,
522 metric.created_at,
523 metric.run_hash.clone(),
524 )
525 })
526 .collect();
527
528 let mut param_idx = 0;
529 let values_clauses: Vec<String> = prepared_data
530 .iter()
531 .map(|_| {
532 let clause = format!(
533 "(${}, ${}, ${}, ${}, ${}, ${}, ${}, ${})",
534 param_idx + 1,
535 param_idx + 2,
536 param_idx + 3,
537 param_idx + 4,
538 param_idx + 5,
539 param_idx + 6,
540 param_idx + 7,
541 param_idx + 8
542 );
543 param_idx += 8;
544 clause
545 })
546 .collect();
547
548 let sql = format!(
549 "INSERT INTO pipeline_metrics (pipeline_id, run_number, timestamp, metric_type, value, metadata_json, created_at, run_hash)
550 VALUES {}
551 ON CONFLICT DO NOTHING",
552 values_clauses.join(", ")
553 );
554
555 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
556 for (
557 pipeline_id,
558 run_number,
559 timestamp,
560 metric_type,
561 value,
562 metadata_json,
563 created_at,
564 run_hash,
565 ) in &prepared_data
566 {
567 query = query
568 .bind(pipeline_id)
569 .bind(run_number)
570 .bind(timestamp)
571 .bind(metric_type)
572 .bind(value)
573 .bind(metadata_json)
574 .bind(created_at)
575 .bind(run_hash);
576 }
577
578 let result = query
579 .execute(&mut *tx)
580 .await
581 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
582
583 total_inserted += result.rows_affected() as usize;
584 }
585
586 tx.commit()
587 .await
588 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
589 }
590 }
591
592 let skipped = metrics_count - total_inserted;
593 if skipped > 0 {
594 tracing::debug!(skipped = skipped, "Skipped duplicate metrics");
595 }
596
597 let elapsed = start.elapsed();
598 tracing::debug!(
599 metrics_count = metrics_count,
600 inserted = total_inserted,
601 skipped = skipped,
602 elapsed_ms = elapsed.as_millis(),
603 "Inserted metrics batch"
604 );
605
606 Ok(total_inserted)
607 }
608
609 pub async fn refresh_record_metrics(
610 &self, pipeline_id: &str, records: &[RecordMetrics],
611 ) -> DomainResult<usize> {
612 if records.is_empty() {
613 return Ok(0);
614 }
615
616 let mut affected_rows = 0;
617
618 match &self.pool {
619 DatabasePool::Sqlite(p) => {
620 let mut tx = p
621 .begin()
622 .await
623 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
624
625 for record in records {
626 let deleted = sqlx::query(
627 "DELETE FROM pipeline_metrics
628 WHERE pipeline_id = ? AND run_number = ?
629 AND (run_hash IS NULL OR run_hash <> ?)",
630 )
631 .bind(pipeline_id)
632 .bind(record.run_number)
633 .bind(&record.run_hash)
634 .execute(&mut *tx)
635 .await
636 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
637 affected_rows += deleted.rows_affected() as usize;
638
639 for metric in record.entries.iter().flatten() {
640 let result = sqlx::query(
641 "INSERT INTO pipeline_metrics (pipeline_id, run_number, timestamp, metric_type, value, metadata_json, created_at, run_hash)
642 SELECT ?, ?, ?, ?, ?, ?, ?, ?
643 WHERE NOT EXISTS (
644 SELECT 1 FROM pipeline_metrics
645 WHERE pipeline_id = ? AND run_number = ? AND metric_type = ?
646 )",
647 )
648 .bind(pipeline_id)
649 .bind(record.run_number)
650 .bind(metric.timestamp.to_rfc3339())
651 .bind(metric.metric_type.as_str())
652 .bind(metric.value)
653 .bind(metadata_json(metric))
654 .bind(metric.created_at.to_rfc3339())
655 .bind(&record.run_hash)
656 .bind(pipeline_id)
657 .bind(record.run_number)
658 .bind(metric.metric_type.as_str())
659 .execute(&mut *tx)
660 .await
661 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
662
663 affected_rows += result.rows_affected() as usize;
664 }
665 }
666
667 tx.commit()
668 .await
669 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
670 }
671 DatabasePool::Postgres(p) => {
672 let mut tx = p
673 .begin()
674 .await
675 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
676
677 sqlx::query("SELECT pg_advisory_xact_lock($1, hashtext($2))")
678 .bind(METRICS_LOCK_NAMESPACE)
679 .bind(pipeline_id)
680 .execute(&mut *tx)
681 .await
682 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
683
684 for record in records {
685 let deleted = sqlx::query(
686 "DELETE FROM pipeline_metrics
687 WHERE pipeline_id = $1 AND run_number = $2
688 AND (run_hash IS NULL OR run_hash <> $3)",
689 )
690 .bind(pipeline_id)
691 .bind(record.run_number)
692 .bind(&record.run_hash)
693 .execute(&mut *tx)
694 .await
695 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
696 affected_rows += deleted.rows_affected() as usize;
697
698 for metric in record.entries.iter().flatten() {
699 let result = sqlx::query(
700 "INSERT INTO pipeline_metrics (pipeline_id, run_number, timestamp, metric_type, value, metadata_json, created_at, run_hash)
701 SELECT $1, $2, $3, $4, $5, $6, $7, $8
702 WHERE NOT EXISTS (
703 SELECT 1 FROM pipeline_metrics
704 WHERE pipeline_id = $1 AND run_number = $2 AND metric_type = $4
705 )",
706 )
707 .bind(pipeline_id)
708 .bind(record.run_number)
709 .bind(metric.timestamp)
710 .bind(metric.metric_type.as_str())
711 .bind(metric.value)
712 .bind(metadata_json(metric))
713 .bind(metric.created_at)
714 .bind(&record.run_hash)
715 .execute(&mut *tx)
716 .await
717 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
718
719 affected_rows += result.rows_affected() as usize;
720 }
721 }
722
723 tx.commit()
724 .await
725 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
726 }
727 }
728
729 tracing::debug!(
730 pipeline_id = %pipeline_id,
731 records = records.len(),
732 affected_rows,
733 "Refreshed record metrics"
734 );
735
736 Ok(affected_rows)
737 }
738
739 pub async fn query_metrics(&self, query: MetricsQuery) -> DomainResult<Vec<MetricEntry>> {
740 let mut param_idx = 0;
741 let base_select = match &self.pool {
742 DatabasePool::Sqlite(_) => "SELECT id, pipeline_id, run_number, timestamp, metric_type, value, metadata_json, created_at, run_hash FROM pipeline_metrics WHERE 1=1",
743 DatabasePool::Postgres(_) => "SELECT id, pipeline_id, run_number::BIGINT, timestamp, metric_type, value, metadata_json, created_at, run_hash FROM pipeline_metrics WHERE 1=1",
744 };
745 let mut sql = String::from(base_select);
746
747 let mut string_params: Vec<(usize, String)> = Vec::new();
748 let mut timestamp_params: Vec<(usize, DateTime<Utc>)> = Vec::new();
749
750 if let Some(pipeline_id) = &query.pipeline_id {
751 param_idx += 1;
752 sql.push_str(&format!(
753 " AND pipeline_id = {}",
754 self.placeholder(param_idx)
755 ));
756 string_params.push((param_idx, pipeline_id.clone()));
757 }
758
759 if let Some(metric_type) = &query.metric_type {
760 param_idx += 1;
761 sql.push_str(&format!(
762 " AND metric_type = {}",
763 self.placeholder(param_idx)
764 ));
765 string_params.push((param_idx, metric_type.as_str().to_string()));
766 }
767
768 if let Some(start_date) = &query.start_date {
769 param_idx += 1;
770 sql.push_str(&format!(
771 " AND timestamp >= {}",
772 self.placeholder(param_idx)
773 ));
774 timestamp_params.push((param_idx, *start_date));
775 }
776
777 if let Some(end_date) = &query.end_date {
778 param_idx += 1;
779 sql.push_str(&format!(
780 " AND timestamp <= {}",
781 self.placeholder(param_idx)
782 ));
783 timestamp_params.push((param_idx, *end_date));
784 }
785
786 sql.push_str(" ORDER BY timestamp DESC");
787
788 if let Some(limit) = query.limit {
789 sql.push_str(&format!(" LIMIT {}", limit));
790 }
791
792 match &self.pool {
793 DatabasePool::Sqlite(p) => {
794 let mut query_builder = sqlx::query(sqlx::AssertSqlSafe(sql));
795 let mut all_params: Vec<(usize, String)> = string_params;
796 for (idx, ts) in timestamp_params {
797 all_params.push((idx, ts.to_rfc3339()));
798 }
799 all_params.sort_by_key(|(idx, _)| *idx);
800 for (_, value) in all_params {
801 query_builder = query_builder.bind(value);
802 }
803
804 let rows = query_builder
805 .fetch_all(p)
806 .await
807 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
808
809 rows.iter()
810 .map(|row| self.metric_entry_from_sqlite_row(row))
811 .collect::<DomainResult<Vec<_>>>()
812 }
813 DatabasePool::Postgres(p) => {
814 enum BindValue {
815 Str(String),
816 Timestamp(DateTime<Utc>),
817 }
818 let mut all_params: Vec<(usize, BindValue)> = Vec::new();
819 for (idx, s) in string_params {
820 all_params.push((idx, BindValue::Str(s)));
821 }
822 for (idx, ts) in timestamp_params {
823 all_params.push((idx, BindValue::Timestamp(ts)));
824 }
825 all_params.sort_by_key(|(idx, _)| *idx);
826
827 let mut query_builder = sqlx::query(sqlx::AssertSqlSafe(sql));
828 for (_, value) in all_params {
829 match value {
830 BindValue::Str(s) => query_builder = query_builder.bind(s),
831 BindValue::Timestamp(ts) => query_builder = query_builder.bind(ts),
832 }
833 }
834
835 let rows = query_builder
836 .fetch_all(p)
837 .await
838 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
839
840 rows.iter()
841 .map(|row| self.metric_entry_from_postgres_row(row))
842 .collect::<DomainResult<Vec<_>>>()
843 }
844 }
845 }
846
847 pub async fn query_aggregated_metrics(
848 &self, query: MetricsQuery,
849 ) -> DomainResult<AggregatedMetrics> {
850 let metric_type = query
851 .metric_type
852 .ok_or_else(|| DomainError::InvalidConfig("metric_type is required".to_string()))?;
853 let aggregation_period = query.aggregation_period.ok_or_else(|| {
854 DomainError::InvalidConfig("aggregation_period is required".to_string())
855 })?;
856 let aggregation_type = query.aggregation_type.unwrap_or(AggregationType::Avg);
857
858 let period_select = match (&self.pool, aggregation_period) {
859 (DatabasePool::Sqlite(_), AggregationPeriod::Hourly) => {
860 "datetime(strftime('%Y-%m-%d %H:00:00', timestamp))"
861 }
862 (DatabasePool::Sqlite(_), AggregationPeriod::Daily) => "date(timestamp)",
863 (DatabasePool::Sqlite(_), AggregationPeriod::Weekly) => {
864 "date(timestamp, 'weekday 0', '-6 days')"
865 }
866 (DatabasePool::Sqlite(_), AggregationPeriod::Monthly) => {
867 "date(timestamp, 'start of month')"
868 }
869 (DatabasePool::Postgres(_), AggregationPeriod::Hourly) => {
870 "date_trunc('hour', timestamp)"
871 }
872 (DatabasePool::Postgres(_), AggregationPeriod::Daily) => "date_trunc('day', timestamp)",
873 (DatabasePool::Postgres(_), AggregationPeriod::Weekly) => {
874 "date_trunc('week', timestamp)"
875 }
876 (DatabasePool::Postgres(_), AggregationPeriod::Monthly) => {
877 "date_trunc('month', timestamp)"
878 }
879 };
880
881 let use_percentile = matches!(
882 aggregation_type,
883 AggregationType::P95 | AggregationType::P99
884 );
885
886 let metrics = if use_percentile {
887 self.calculate_percentile_metrics(
888 &query,
889 metric_type,
890 aggregation_period,
891 aggregation_type,
892 period_select,
893 )
894 .await?
895 } else {
896 let aggregation_select = match aggregation_type {
897 AggregationType::Avg => "AVG(value)",
898 AggregationType::Sum => "SUM(value)",
899 AggregationType::Min => "MIN(value)",
900 AggregationType::Max => "MAX(value)",
901 _ => unreachable!(),
902 };
903
904 let mut param_idx = 0;
905 let mut sql = format!(
906 "SELECT
907 {} as period,
908 {} as agg_value,
909 COUNT(*) as count,
910 MIN(value) as min_value,
911 MAX(value) as max_value,
912 AVG(value) as avg_value
913 FROM pipeline_metrics
914 WHERE metric_type = {}",
915 period_select,
916 aggregation_select,
917 {
918 param_idx += 1;
919 self.placeholder(param_idx)
920 }
921 );
922
923 let mut string_params: Vec<(usize, String)> =
924 vec![(1, metric_type.as_str().to_string())];
925 let mut timestamp_params: Vec<(usize, DateTime<Utc>)> = Vec::new();
926
927 if let Some(pipeline_id) = &query.pipeline_id {
928 param_idx += 1;
929 sql.push_str(&format!(
930 " AND pipeline_id = {}",
931 self.placeholder(param_idx)
932 ));
933 string_params.push((param_idx, pipeline_id.clone()));
934 }
935
936 if let Some(start_date) = &query.start_date {
937 param_idx += 1;
938 sql.push_str(&format!(
939 " AND timestamp >= {}",
940 self.placeholder(param_idx)
941 ));
942 timestamp_params.push((param_idx, *start_date));
943 }
944
945 if let Some(end_date) = &query.end_date {
946 param_idx += 1;
947 sql.push_str(&format!(
948 " AND timestamp <= {}",
949 self.placeholder(param_idx)
950 ));
951 timestamp_params.push((param_idx, *end_date));
952 }
953
954 sql.push_str(" GROUP BY period ORDER BY period ASC");
955
956 if let Some(limit) = query.limit {
957 sql.push_str(&format!(" LIMIT {}", limit));
958 }
959
960 match &self.pool {
961 DatabasePool::Sqlite(p) => {
962 let mut query_builder = sqlx::query(sqlx::AssertSqlSafe(sql));
963 let mut all_params: Vec<(usize, String)> = string_params;
964 for (idx, ts) in timestamp_params {
965 all_params.push((idx, ts.to_rfc3339()));
966 }
967 all_params.sort_by_key(|(idx, _)| *idx);
968 for (_, value) in all_params {
969 query_builder = query_builder.bind(value);
970 }
971
972 let rows = query_builder
973 .fetch_all(p)
974 .await
975 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
976
977 rows.iter()
978 .map(|row| {
979 let period_str: String = row
980 .try_get(0)
981 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
982 let timestamp =
983 self.parse_period_timestamp(&period_str, aggregation_period);
984
985 Ok(AggregatedMetric {
986 timestamp,
987 value: row
988 .try_get::<f64, _>(1)
989 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
990 count: row
991 .try_get(2)
992 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
993 min: row.try_get(3).ok(),
994 max: row.try_get(4).ok(),
995 avg: row
996 .try_get(5)
997 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
998 })
999 })
1000 .collect::<DomainResult<Vec<_>>>()?
1001 }
1002 DatabasePool::Postgres(p) => {
1003 enum BindValue {
1004 Str(String),
1005 Timestamp(DateTime<Utc>),
1006 }
1007 let mut all_params: Vec<(usize, BindValue)> = Vec::new();
1008 for (idx, s) in string_params {
1009 all_params.push((idx, BindValue::Str(s)));
1010 }
1011 for (idx, ts) in timestamp_params {
1012 all_params.push((idx, BindValue::Timestamp(ts)));
1013 }
1014 all_params.sort_by_key(|(idx, _)| *idx);
1015
1016 let mut query_builder = sqlx::query(sqlx::AssertSqlSafe(sql));
1017 for (_, value) in all_params {
1018 match value {
1019 BindValue::Str(s) => query_builder = query_builder.bind(s),
1020 BindValue::Timestamp(ts) => query_builder = query_builder.bind(ts),
1021 }
1022 }
1023
1024 let rows = query_builder
1025 .fetch_all(p)
1026 .await
1027 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1028
1029 rows.iter()
1030 .map(|row| {
1031 let timestamp: DateTime<Utc> = row
1032 .try_get(0)
1033 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1034
1035 Ok(AggregatedMetric {
1036 timestamp,
1037 value: row
1038 .try_get::<f64, _>(1)
1039 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1040 count: row
1041 .try_get(2)
1042 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1043 min: row.try_get(3).ok(),
1044 max: row.try_get(4).ok(),
1045 avg: row
1046 .try_get(5)
1047 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1048 })
1049 })
1050 .collect::<DomainResult<Vec<_>>>()?
1051 }
1052 }
1053 };
1054
1055 let total_count = metrics.len();
1056
1057 Ok(AggregatedMetrics {
1058 metrics,
1059 total_count,
1060 metric_type,
1061 aggregation_period,
1062 })
1063 }
1064
1065 async fn calculate_percentile_metrics(
1066 &self, query: &MetricsQuery, metric_type: MetricType,
1067 aggregation_period: AggregationPeriod, aggregation_type: AggregationType,
1068 period_select: &str,
1069 ) -> DomainResult<Vec<AggregatedMetric>> {
1070 let percentile = if matches!(aggregation_type, AggregationType::P95) {
1071 0.95
1072 } else {
1073 0.99
1074 };
1075
1076 let mut param_idx = 0;
1077 let mut sql = format!(
1078 "SELECT {} as period, value FROM pipeline_metrics WHERE metric_type = {}",
1079 period_select,
1080 {
1081 param_idx += 1;
1082 self.placeholder(param_idx)
1083 }
1084 );
1085
1086 let mut string_params: Vec<(usize, String)> = vec![(1, metric_type.as_str().to_string())];
1087 let mut timestamp_params: Vec<(usize, DateTime<Utc>)> = Vec::new();
1088
1089 if let Some(pipeline_id) = &query.pipeline_id {
1090 param_idx += 1;
1091 sql.push_str(&format!(
1092 " AND pipeline_id = {}",
1093 self.placeholder(param_idx)
1094 ));
1095 string_params.push((param_idx, pipeline_id.clone()));
1096 }
1097
1098 if let Some(start_date) = &query.start_date {
1099 param_idx += 1;
1100 sql.push_str(&format!(
1101 " AND timestamp >= {}",
1102 self.placeholder(param_idx)
1103 ));
1104 timestamp_params.push((param_idx, *start_date));
1105 }
1106
1107 if let Some(end_date) = &query.end_date {
1108 param_idx += 1;
1109 sql.push_str(&format!(
1110 " AND timestamp <= {}",
1111 self.placeholder(param_idx)
1112 ));
1113 timestamp_params.push((param_idx, *end_date));
1114 }
1115
1116 sql.push_str(" ORDER BY period, value");
1117
1118 let data: Vec<(String, f64)> = match &self.pool {
1119 DatabasePool::Sqlite(p) => {
1120 let mut query_builder = sqlx::query(sqlx::AssertSqlSafe(sql));
1121 let mut all_params: Vec<(usize, String)> = string_params;
1122 for (idx, ts) in timestamp_params {
1123 all_params.push((idx, ts.to_rfc3339()));
1124 }
1125 all_params.sort_by_key(|(idx, _)| *idx);
1126 for (_, value) in all_params {
1127 query_builder = query_builder.bind(value);
1128 }
1129
1130 let rows = query_builder
1131 .fetch_all(p)
1132 .await
1133 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1134
1135 rows.iter()
1136 .map(|row| {
1137 let period: String = row
1138 .try_get(0)
1139 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1140 let value: f64 = row
1141 .try_get(1)
1142 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1143 Ok((period, value))
1144 })
1145 .collect::<DomainResult<Vec<_>>>()?
1146 }
1147 DatabasePool::Postgres(p) => {
1148 enum BindValue {
1149 Str(String),
1150 Timestamp(DateTime<Utc>),
1151 }
1152 let mut all_params: Vec<(usize, BindValue)> = Vec::new();
1153 for (idx, s) in string_params {
1154 all_params.push((idx, BindValue::Str(s)));
1155 }
1156 for (idx, ts) in timestamp_params {
1157 all_params.push((idx, BindValue::Timestamp(ts)));
1158 }
1159 all_params.sort_by_key(|(idx, _)| *idx);
1160
1161 let mut query_builder = sqlx::query(sqlx::AssertSqlSafe(sql));
1162 for (_, value) in all_params {
1163 match value {
1164 BindValue::Str(s) => query_builder = query_builder.bind(s),
1165 BindValue::Timestamp(ts) => query_builder = query_builder.bind(ts),
1166 }
1167 }
1168
1169 let rows = query_builder
1170 .fetch_all(p)
1171 .await
1172 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1173
1174 rows.iter()
1175 .map(|row| {
1176 let period: DateTime<Utc> = row
1177 .try_get(0)
1178 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1179 let value: f64 = row
1180 .try_get(1)
1181 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1182 Ok((period.to_rfc3339(), value))
1183 })
1184 .collect::<DomainResult<Vec<_>>>()?
1185 }
1186 };
1187
1188 let mut period_values: std::collections::BTreeMap<String, Vec<f64>> =
1189 std::collections::BTreeMap::new();
1190
1191 for (period, value) in data {
1192 if !value.is_nan() && value.is_finite() {
1193 period_values.entry(period).or_default().push(value);
1194 }
1195 }
1196
1197 let metrics: Vec<AggregatedMetric> = period_values
1198 .into_iter()
1199 .filter_map(|(period_str, mut values)| {
1200 if values.is_empty() {
1201 return None;
1202 }
1203
1204 values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
1205
1206 let count = values.len();
1207 let min = values.first().copied();
1208 let max = values.last().copied();
1209 let sum: f64 = values.iter().sum();
1210 let avg = sum / count as f64;
1211
1212 let percentile_value = self.calculate_percentile(&values, percentile);
1213
1214 let timestamp = self.parse_period_timestamp(&period_str, aggregation_period);
1215
1216 Some(AggregatedMetric {
1217 timestamp,
1218 value: percentile_value,
1219 count: count as i64,
1220 min,
1221 max,
1222 avg,
1223 })
1224 })
1225 .collect();
1226
1227 let mut sorted_metrics = metrics;
1228 sorted_metrics.sort_by_key(|a| a.timestamp);
1229
1230 if let Some(limit) = query.limit {
1231 sorted_metrics.truncate(limit);
1232 }
1233
1234 Ok(sorted_metrics)
1235 }
1236
1237 fn calculate_percentile(&self, sorted_values: &[f64], percentile: f64) -> f64 {
1238 if sorted_values.is_empty() {
1239 return 0.0;
1240 }
1241
1242 if sorted_values.len() == 1 {
1243 return sorted_values[0];
1244 }
1245
1246 let position = percentile * (sorted_values.len() - 1) as f64;
1247 let lower_index = position.floor() as usize;
1248 let upper_index = position.ceil() as usize;
1249
1250 if lower_index == upper_index {
1251 sorted_values[lower_index]
1252 } else {
1253 let lower_value = sorted_values[lower_index];
1254 let upper_value = sorted_values[upper_index];
1255 let fraction = position - lower_index as f64;
1256 lower_value + (upper_value - lower_value) * fraction
1257 }
1258 }
1259
1260 fn parse_period_timestamp(
1261 &self, period_str: &str, aggregation_period: AggregationPeriod,
1262 ) -> DateTime<Utc> {
1263 if let Ok(dt) = DateTime::parse_from_rfc3339(period_str) {
1264 return dt.with_timezone(&Utc);
1265 }
1266
1267 if aggregation_period == AggregationPeriod::Weekly {
1268 DateTime::parse_from_str(
1269 &format!("{} 00:00:00 +0000", period_str),
1270 "%Y-%m-%d %H:%M:%S %z",
1271 )
1272 .map(|dt| dt.with_timezone(&Utc))
1273 .unwrap_or_else(|_| Utc::now())
1274 } else {
1275 let datetime_str = if period_str.contains(' ') {
1276 format!("{} +0000", period_str)
1277 } else {
1278 format!("{} 00:00:00 +0000", period_str)
1279 };
1280
1281 DateTime::parse_from_str(&datetime_str, "%Y-%m-%d %H:%M:%S %z")
1282 .map(|dt| dt.with_timezone(&Utc))
1283 .unwrap_or_else(|_| Utc::now())
1284 }
1285 }
1286
1287 pub async fn delete_old_metrics(&self, pipeline_id: Option<&str>) -> DomainResult<usize> {
1288 let start = std::time::Instant::now();
1289 let global_config = self.get_global_config().await?;
1290 let mut total_deleted = 0;
1291
1292 if let Some(pid) = pipeline_id {
1293 let config = self.get_pipeline_config(pid).await?;
1294 let retention_days = config
1295 .as_ref()
1296 .map(|c| c.retention_days)
1297 .unwrap_or(global_config.default_retention_days);
1298
1299 let cutoff_date = Utc::now() - Duration::days(retention_days);
1300
1301 let deleted = match &self.pool {
1302 DatabasePool::Sqlite(p) => {
1303 let result = sqlx::query(
1304 "DELETE FROM pipeline_metrics WHERE pipeline_id = ? AND datetime(timestamp) < datetime(?)",
1305 )
1306 .bind(pid)
1307 .bind(cutoff_date.to_rfc3339())
1308 .execute(p)
1309 .await
1310 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1311 result.rows_affected() as usize
1312 }
1313 DatabasePool::Postgres(p) => {
1314 let result = sqlx::query(
1315 "DELETE FROM pipeline_metrics WHERE pipeline_id = $1 AND timestamp < $2",
1316 )
1317 .bind(pid)
1318 .bind(cutoff_date)
1319 .execute(p)
1320 .await
1321 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1322 result.rows_affected() as usize
1323 }
1324 };
1325
1326 total_deleted += deleted;
1327 } else {
1328 use std::collections::HashMap;
1329
1330 let pipeline_ids = self.list_pipeline_ids().await?;
1331
1332 if pipeline_ids.is_empty() {
1333 return Ok(0);
1334 }
1335
1336 let configs: HashMap<String, i64> = match &self.pool {
1337 DatabasePool::Sqlite(p) => sqlx::query_as::<_, (String, i64)>(
1338 "SELECT pipeline_id, retention_days FROM metrics_config",
1339 )
1340 .fetch_all(p)
1341 .await
1342 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1343 .into_iter()
1344 .collect(),
1345 DatabasePool::Postgres(p) => sqlx::query_as::<_, (String, i64)>(
1346 "SELECT pipeline_id, retention_days::BIGINT FROM metrics_config",
1347 )
1348 .fetch_all(p)
1349 .await
1350 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1351 .into_iter()
1352 .collect(),
1353 };
1354
1355 let mut pipelines_by_retention: HashMap<i64, Vec<String>> = HashMap::new();
1356
1357 for pid in pipeline_ids {
1358 let retention_days = configs
1359 .get(&pid)
1360 .copied()
1361 .unwrap_or(global_config.default_retention_days);
1362
1363 pipelines_by_retention
1364 .entry(retention_days)
1365 .or_default()
1366 .push(pid);
1367 }
1368
1369 for (retention_days, pids) in pipelines_by_retention {
1370 let cutoff_date = Utc::now() - Duration::days(retention_days);
1371
1372 match &self.pool {
1373 DatabasePool::Sqlite(p) => {
1374 let placeholders = pids.iter().map(|_| "?").collect::<Vec<_>>().join(", ");
1375 let sql = format!(
1376 "DELETE FROM pipeline_metrics WHERE pipeline_id IN ({}) AND datetime(timestamp) < datetime(?)",
1377 placeholders
1378 );
1379
1380 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
1381 for pid in &pids {
1382 query = query.bind(pid);
1383 }
1384 query = query.bind(cutoff_date.to_rfc3339());
1385
1386 let result = query
1387 .execute(p)
1388 .await
1389 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1390
1391 total_deleted += result.rows_affected() as usize;
1392 }
1393 DatabasePool::Postgres(p) => {
1394 let mut param_idx = 0;
1395 let placeholders = pids
1396 .iter()
1397 .map(|_| {
1398 param_idx += 1;
1399 format!("${}", param_idx)
1400 })
1401 .collect::<Vec<_>>()
1402 .join(", ");
1403 param_idx += 1;
1404 let sql = format!(
1405 "DELETE FROM pipeline_metrics WHERE pipeline_id IN ({}) AND timestamp < ${}",
1406 placeholders, param_idx
1407 );
1408
1409 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
1410 for pid in &pids {
1411 query = query.bind(pid);
1412 }
1413 query = query.bind(cutoff_date);
1414
1415 let result = query
1416 .execute(p)
1417 .await
1418 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1419
1420 total_deleted += result.rows_affected() as usize;
1421 }
1422 }
1423 }
1424 }
1425
1426 match &self.pool {
1427 DatabasePool::Sqlite(p) => {
1428 let now = Utc::now().to_rfc3339();
1429 sqlx::query(
1430 "UPDATE metrics_storage_info SET last_cleanup_at = ?, updated_at = ? WHERE id = 1",
1431 )
1432 .bind(&now)
1433 .bind(&now)
1434 .execute(p)
1435 .await
1436 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1437 }
1438 DatabasePool::Postgres(p) => {
1439 let now = Utc::now();
1440 sqlx::query(
1441 "UPDATE metrics_storage_info SET last_cleanup_at = $1, updated_at = $2 WHERE id = 1",
1442 )
1443 .bind(now)
1444 .bind(now)
1445 .execute(p)
1446 .await
1447 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1448 }
1449 }
1450
1451 let elapsed = start.elapsed();
1452 tracing::debug!(
1453 pipeline_id = pipeline_id,
1454 total_deleted = total_deleted,
1455 elapsed_ms = elapsed.as_millis(),
1456 "Deleted old metrics"
1457 );
1458
1459 Ok(total_deleted)
1460 }
1461
1462 pub async fn flush_metrics(
1463 &self, pipeline_id: Option<&str>, skip_vacuum: bool,
1464 ) -> DomainResult<usize> {
1465 let deleted_count = match &self.pool {
1466 DatabasePool::Sqlite(p) => {
1467 let result = if let Some(pid) = pipeline_id {
1468 sqlx::query("DELETE FROM pipeline_metrics WHERE pipeline_id = ?")
1469 .bind(pid)
1470 .execute(p)
1471 .await
1472 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1473 } else {
1474 sqlx::query("DELETE FROM pipeline_metrics")
1475 .execute(p)
1476 .await
1477 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1478 };
1479 result.rows_affected() as usize
1480 }
1481 DatabasePool::Postgres(p) => {
1482 let result = if let Some(pid) = pipeline_id {
1483 sqlx::query("DELETE FROM pipeline_metrics WHERE pipeline_id = $1")
1484 .bind(pid)
1485 .execute(p)
1486 .await
1487 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1488 } else {
1489 sqlx::query("DELETE FROM pipeline_metrics")
1490 .execute(p)
1491 .await
1492 .map_err(|e| DomainError::DatabaseError(e.to_string()))?
1493 };
1494 result.rows_affected() as usize
1495 }
1496 };
1497
1498 if let DatabasePool::Sqlite(p) = &self.pool {
1499 if deleted_count > 0 && !skip_vacuum {
1500 tracing::debug!(
1501 deleted = deleted_count,
1502 "Running VACUUM after flushing metrics"
1503 );
1504 sqlx::query("VACUUM")
1505 .execute(p)
1506 .await
1507 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1508 tracing::debug!("VACUUM complete");
1509 }
1510 }
1511
1512 Ok(deleted_count)
1513 }
1514
1515 pub async fn calculate_storage_stats(&self) -> DomainResult<MetricsStats> {
1516 let total_metrics_count: i64 = match &self.pool {
1517 DatabasePool::Sqlite(p) => sqlx::query_scalar("SELECT COUNT(*) FROM pipeline_metrics")
1518 .fetch_one(p)
1519 .await
1520 .unwrap_or(0),
1521 DatabasePool::Postgres(p) => {
1522 sqlx::query_scalar("SELECT COUNT(*) FROM pipeline_metrics")
1523 .fetch_one(p)
1524 .await
1525 .unwrap_or(0)
1526 }
1527 };
1528
1529 const ESTIMATED_ROW_SIZE_BYTES: i64 = 250;
1530 const OVERHEAD_MULTIPLIER: f64 = 1.2;
1531
1532 let estimated_size_bytes =
1533 ((total_metrics_count * ESTIMATED_ROW_SIZE_BYTES) as f64 * OVERHEAD_MULTIPLIER) as i64;
1534 let estimated_size_mb = estimated_size_bytes as f64 / 1024.0 / 1024.0;
1535
1536 let last_cleanup_at: Option<DateTime<Utc>> = match &self.pool {
1537 DatabasePool::Sqlite(p) => sqlx::query_scalar::<_, Option<String>>(
1538 "SELECT last_cleanup_at FROM metrics_storage_info WHERE id = 1",
1539 )
1540 .fetch_optional(p)
1541 .await
1542 .ok()
1543 .flatten()
1544 .flatten()
1545 .and_then(|s| {
1546 DateTime::parse_from_rfc3339(&s)
1547 .ok()
1548 .map(|dt| dt.with_timezone(&Utc))
1549 }),
1550 DatabasePool::Postgres(p) => sqlx::query_scalar::<_, Option<DateTime<Utc>>>(
1551 "SELECT last_cleanup_at FROM metrics_storage_info WHERE id = 1",
1552 )
1553 .fetch_optional(p)
1554 .await
1555 .ok()
1556 .flatten()
1557 .flatten(),
1558 };
1559
1560 let by_pipeline = match &self.pool {
1561 DatabasePool::Sqlite(p) => {
1562 let rows = sqlx::query(
1563 "SELECT pm.pipeline_id, COUNT(*) as count, MIN(pm.timestamp) as oldest, MAX(pm.timestamp) as newest
1564 FROM pipeline_metrics pm
1565 GROUP BY pm.pipeline_id",
1566 )
1567 .fetch_all(p)
1568 .await
1569 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1570
1571 rows.iter()
1572 .map(|row| self.pipeline_metrics_stats_from_sqlite_row(row))
1573 .collect::<DomainResult<Vec<_>>>()?
1574 }
1575 DatabasePool::Postgres(p) => {
1576 let rows = sqlx::query(
1577 "SELECT pm.pipeline_id, COUNT(*) as count, MIN(pm.timestamp) as oldest, MAX(pm.timestamp) as newest
1578 FROM pipeline_metrics pm
1579 GROUP BY pm.pipeline_id",
1580 )
1581 .fetch_all(p)
1582 .await
1583 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1584
1585 rows.iter()
1586 .map(|row| self.pipeline_metrics_stats_from_postgres_row(row))
1587 .collect::<DomainResult<Vec<_>>>()?
1588 }
1589 };
1590
1591 Ok(MetricsStats {
1592 total_metrics_count,
1593 estimated_size_bytes,
1594 estimated_size_mb,
1595 last_cleanup_at,
1596 updated_at: Utc::now(),
1597 by_pipeline,
1598 })
1599 }
1600
1601 pub async fn get_storage_stats(&self) -> DomainResult<MetricsStats> {
1602 self.calculate_storage_stats().await
1603 }
1604
1605 fn metrics_config_from_sqlite_row(
1606 &self, row: &sqlx::sqlite::SqliteRow,
1607 ) -> DomainResult<MetricsConfig> {
1608 let enabled: i64 = row
1609 .try_get(1)
1610 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1611 let created_at_str: String = row
1612 .try_get(3)
1613 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1614 let updated_at_str: String = row
1615 .try_get(4)
1616 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1617
1618 Ok(MetricsConfig {
1619 pipeline_id: row
1620 .try_get(0)
1621 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1622 enabled: enabled != 0,
1623 retention_days: row
1624 .try_get(2)
1625 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1626 created_at: DateTime::parse_from_rfc3339(&created_at_str)
1627 .map(|dt| dt.with_timezone(&Utc))
1628 .unwrap_or_else(|_| Utc::now()),
1629 updated_at: DateTime::parse_from_rfc3339(&updated_at_str)
1630 .map(|dt| dt.with_timezone(&Utc))
1631 .unwrap_or_else(|_| Utc::now()),
1632 })
1633 }
1634
1635 fn metrics_config_from_postgres_row(
1636 &self, row: &sqlx::postgres::PgRow,
1637 ) -> DomainResult<MetricsConfig> {
1638 Ok(MetricsConfig {
1639 pipeline_id: row
1640 .try_get(0)
1641 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1642 enabled: row
1643 .try_get(1)
1644 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1645 retention_days: row
1646 .try_get(2)
1647 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1648 created_at: row
1649 .try_get(3)
1650 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1651 updated_at: row
1652 .try_get(4)
1653 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1654 })
1655 }
1656
1657 fn metric_entry_from_sqlite_row(
1658 &self, row: &sqlx::sqlite::SqliteRow,
1659 ) -> DomainResult<MetricEntry> {
1660 let metric_type_str: String = row
1661 .try_get(4)
1662 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1663 let metric_type = metric_type_str.parse().unwrap_or(MetricType::RunDuration);
1664
1665 let metadata_str: Option<String> = row
1666 .try_get(6)
1667 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1668 let metadata: Option<serde_json::Value> =
1669 metadata_str.and_then(|s| serde_json::from_str(&s).ok());
1670
1671 let timestamp_str: String = row
1672 .try_get(3)
1673 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1674 let created_at_str: String = row
1675 .try_get(7)
1676 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1677 let run_hash: Option<String> = row.try_get(8).ok();
1678
1679 Ok(MetricEntry {
1680 id: row
1681 .try_get(0)
1682 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1683 pipeline_id: row
1684 .try_get(1)
1685 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1686 run_number: row
1687 .try_get(2)
1688 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1689 timestamp: DateTime::parse_from_rfc3339(×tamp_str)
1690 .map(|dt| dt.with_timezone(&Utc))
1691 .unwrap_or_else(|_| Utc::now()),
1692 metric_type,
1693 value: row
1694 .try_get(5)
1695 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1696 metadata,
1697 created_at: DateTime::parse_from_rfc3339(&created_at_str)
1698 .map(|dt| dt.with_timezone(&Utc))
1699 .unwrap_or_else(|_| Utc::now()),
1700 run_hash,
1701 })
1702 }
1703
1704 fn metric_entry_from_postgres_row(
1705 &self, row: &sqlx::postgres::PgRow,
1706 ) -> DomainResult<MetricEntry> {
1707 let metric_type_str: String = row
1708 .try_get(4)
1709 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1710 let metric_type = metric_type_str.parse().unwrap_or(MetricType::RunDuration);
1711
1712 let metadata_str: Option<String> = row
1713 .try_get(6)
1714 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1715 let metadata: Option<serde_json::Value> =
1716 metadata_str.and_then(|s| serde_json::from_str(&s).ok());
1717
1718 let run_hash: Option<String> = row.try_get(8).ok();
1719
1720 Ok(MetricEntry {
1721 id: row
1722 .try_get(0)
1723 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1724 pipeline_id: row
1725 .try_get(1)
1726 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1727 run_number: row
1728 .try_get(2)
1729 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1730 timestamp: row
1731 .try_get(3)
1732 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1733 metric_type,
1734 value: row
1735 .try_get(5)
1736 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1737 metadata,
1738 created_at: row
1739 .try_get(7)
1740 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1741 run_hash,
1742 })
1743 }
1744
1745 fn pipeline_metrics_stats_from_sqlite_row(
1746 &self, row: &sqlx::sqlite::SqliteRow,
1747 ) -> DomainResult<PipelineMetricsStats> {
1748 let pipeline_id: String = row
1749 .try_get(0)
1750 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1751 let oldest_str: Option<String> = row
1752 .try_get(2)
1753 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1754 let newest_str: Option<String> = row
1755 .try_get(3)
1756 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1757
1758 Ok(PipelineMetricsStats {
1759 pipeline_id: pipeline_id.clone(),
1760 pipeline_name: pipeline_id,
1761 metrics_count: row
1762 .try_get(1)
1763 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1764 oldest_metric: oldest_str.and_then(|s| {
1765 DateTime::parse_from_rfc3339(&s)
1766 .ok()
1767 .map(|dt| dt.with_timezone(&Utc))
1768 }),
1769 newest_metric: newest_str.and_then(|s| {
1770 DateTime::parse_from_rfc3339(&s)
1771 .ok()
1772 .map(|dt| dt.with_timezone(&Utc))
1773 }),
1774 })
1775 }
1776
1777 fn pipeline_metrics_stats_from_postgres_row(
1778 &self, row: &sqlx::postgres::PgRow,
1779 ) -> DomainResult<PipelineMetricsStats> {
1780 let pipeline_id: String = row
1781 .try_get(0)
1782 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1783
1784 Ok(PipelineMetricsStats {
1785 pipeline_id: pipeline_id.clone(),
1786 pipeline_name: pipeline_id,
1787 metrics_count: row
1788 .try_get(1)
1789 .map_err(|e| DomainError::DatabaseError(e.to_string()))?,
1790 oldest_metric: row.try_get(2).ok(),
1791 newest_metric: row.try_get(3).ok(),
1792 })
1793 }
1794
1795 pub async fn export_config(&self) -> DomainResult<MetricsConfigExport> {
1796 let global_config = self.get_global_config().await?;
1797
1798 let pipeline_configs = match &self.pool {
1799 DatabasePool::Sqlite(p) => {
1800 let rows = sqlx::query(
1801 "SELECT pipeline_id, enabled, retention_days, created_at, updated_at FROM metrics_config",
1802 )
1803 .fetch_all(p)
1804 .await
1805 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1806
1807 rows.iter()
1808 .map(|row| self.metrics_config_from_sqlite_row(row))
1809 .collect::<DomainResult<Vec<_>>>()?
1810 }
1811 DatabasePool::Postgres(p) => {
1812 let rows = sqlx::query(
1813 "SELECT pipeline_id, enabled, retention_days::BIGINT, created_at, updated_at FROM metrics_config",
1814 )
1815 .fetch_all(p)
1816 .await
1817 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1818
1819 rows.iter()
1820 .map(|row| self.metrics_config_from_postgres_row(row))
1821 .collect::<DomainResult<Vec<_>>>()?
1822 }
1823 };
1824
1825 Ok(MetricsConfigExport {
1826 global_config,
1827 pipeline_configs,
1828 })
1829 }
1830
1831 pub async fn import_config(&self, export: &MetricsConfigExport) -> DomainResult<()> {
1832 let start = std::time::Instant::now();
1833
1834 self.update_global_config(
1835 export.global_config.enabled,
1836 export.global_config.default_retention_days,
1837 )
1838 .await?;
1839
1840 if export.pipeline_configs.is_empty() {
1841 return Ok(());
1842 }
1843
1844 let configs_count = export.pipeline_configs.len();
1845
1846 const BATCH_SIZE: usize = 100;
1847
1848 match &self.pool {
1849 DatabasePool::Sqlite(p) => {
1850 let mut tx = p
1851 .begin()
1852 .await
1853 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1854
1855 for chunk in export.pipeline_configs.chunks(BATCH_SIZE) {
1856 let values_clause = chunk
1857 .iter()
1858 .map(|_| "(?, ?, ?, datetime('now'), datetime('now'))")
1859 .collect::<Vec<_>>()
1860 .join(", ");
1861
1862 let sql = format!(
1863 "INSERT OR REPLACE INTO metrics_config (pipeline_id, enabled, retention_days, created_at, updated_at) VALUES {}",
1864 values_clause
1865 );
1866
1867 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
1868 for config in chunk {
1869 query = query
1870 .bind(&config.pipeline_id)
1871 .bind(config.enabled)
1872 .bind(config.retention_days);
1873 }
1874
1875 query
1876 .execute(&mut *tx)
1877 .await
1878 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1879 }
1880
1881 tx.commit()
1882 .await
1883 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1884 }
1885 DatabasePool::Postgres(p) => {
1886 let mut tx = p
1887 .begin()
1888 .await
1889 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1890
1891 for chunk in export.pipeline_configs.chunks(BATCH_SIZE) {
1892 let mut param_idx = 0;
1893 let values_clauses: Vec<String> = chunk
1894 .iter()
1895 .map(|_| {
1896 let clause = format!(
1897 "(${}, ${}, ${}, NOW(), NOW())",
1898 param_idx + 1,
1899 param_idx + 2,
1900 param_idx + 3
1901 );
1902 param_idx += 3;
1903 clause
1904 })
1905 .collect();
1906
1907 let sql = format!(
1908 "INSERT INTO metrics_config (pipeline_id, enabled, retention_days, created_at, updated_at) VALUES {}
1909 ON CONFLICT (pipeline_id) DO UPDATE SET enabled = EXCLUDED.enabled, retention_days = EXCLUDED.retention_days, updated_at = EXCLUDED.updated_at",
1910 values_clauses.join(", ")
1911 );
1912
1913 let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
1914 for config in chunk {
1915 query = query
1916 .bind(&config.pipeline_id)
1917 .bind(config.enabled)
1918 .bind(config.retention_days);
1919 }
1920
1921 query
1922 .execute(&mut *tx)
1923 .await
1924 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1925 }
1926
1927 tx.commit()
1928 .await
1929 .map_err(|e| DomainError::DatabaseError(e.to_string()))?;
1930 }
1931 }
1932
1933 let elapsed = start.elapsed();
1934 tracing::debug!(
1935 configs_count = configs_count,
1936 elapsed_ms = elapsed.as_millis(),
1937 "Imported metrics config (batch upsert)"
1938 );
1939
1940 Ok(())
1941 }
1942}