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