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