mockforge_intelligence/handlers/
forecasting.rs

1//! HTTP handlers for API change forecasting
2//!
3//! This module provides endpoints for querying and managing forecasts.
4
5use axum::{
6    extract::{Path, Query, State},
7    http::StatusCode,
8    response::Json,
9};
10use mockforge_contracts::contract_drift::forecasting::{ChangeForecast, Forecaster};
11use serde::{Deserialize, Serialize};
12use std::sync::Arc;
13
14#[cfg(feature = "database")]
15use chrono::{DateTime, Utc};
16#[cfg(feature = "database")]
17use mockforge_contracts::contract_drift::forecasting::SeasonalPattern;
18#[cfg(feature = "database")]
19use uuid::Uuid;
20
21#[cfg(feature = "database")]
22use crate::database::Database;
23
24/// Helper function to map database row to ChangeForecast
25#[cfg(feature = "database")]
26fn map_row_to_change_forecast(row: &sqlx::postgres::PgRow) -> Result<ChangeForecast, sqlx::Error> {
27    use sqlx::Row;
28
29    let service_id: Option<String> = row.try_get("service_id")?;
30    let service_name: Option<String> = row.try_get("service_name")?;
31    let endpoint: String = row.try_get("endpoint")?;
32    let method: String = row.try_get("method")?;
33    let forecast_window_days: i32 = row.try_get("forecast_window_days")?;
34    let predicted_change_probability: f64 = row.try_get("predicted_change_probability")?;
35    let predicted_break_probability: f64 = row.try_get("predicted_break_probability")?;
36    let next_expected_change_date: Option<DateTime<Utc>> =
37        row.try_get("next_expected_change_date")?;
38    let next_expected_break_date: Option<DateTime<Utc>> =
39        row.try_get("next_expected_break_date")?;
40    let volatility_score: f64 = row.try_get("volatility_score")?;
41    let confidence: f64 = row.try_get("confidence")?;
42    let seasonal_patterns_json: serde_json::Value =
43        row.try_get("seasonal_patterns").unwrap_or_default();
44    let predicted_at: DateTime<Utc> = row.try_get("predicted_at")?;
45    let expires_at: DateTime<Utc> = row.try_get("expires_at")?;
46
47    // Parse seasonal patterns from JSONB
48    let seasonal_patterns: Vec<SeasonalPattern> = if seasonal_patterns_json.is_array() {
49        serde_json::from_value(seasonal_patterns_json).unwrap_or_default()
50    } else {
51        Vec::new()
52    };
53
54    Ok(ChangeForecast {
55        service_id,
56        service_name,
57        endpoint,
58        method,
59        forecast_window_days: forecast_window_days as u32,
60        predicted_change_probability,
61        predicted_break_probability,
62        next_expected_change_date,
63        next_expected_break_date,
64        volatility_score,
65        confidence,
66        seasonal_patterns,
67        predicted_at,
68        expires_at,
69    })
70}
71
72/// State for forecasting handlers
73#[derive(Clone)]
74pub struct ForecastingState {
75    /// Forecaster engine
76    pub forecaster: Arc<Forecaster>,
77    /// Database connection (optional)
78    #[cfg(feature = "database")]
79    pub database: Option<Database>,
80}
81
82impl ForecastingState {
83    /// Create handler state without a persistence connection.
84    ///
85    /// This constructor works when dependency feature unification enables database support.
86    pub fn new(forecaster: Arc<Forecaster>) -> Self {
87        Self {
88            forecaster,
89            #[cfg(feature = "database")]
90            database: None,
91        }
92    }
93
94    /// Attach the optional persistence connection used by these handlers.
95    #[cfg(feature = "database")]
96    pub fn with_database(mut self, database: Option<Database>) -> Self {
97        self.database = database;
98        self
99    }
100}
101
102/// Query parameters for listing forecasts
103#[derive(Debug, Deserialize)]
104pub struct ListForecastsQuery {
105    /// Workspace ID filter
106    pub workspace_id: Option<String>,
107    /// Service ID filter
108    pub service_id: Option<String>,
109    /// Endpoint filter
110    pub endpoint: Option<String>,
111    /// Method filter
112    pub method: Option<String>,
113    /// Forecast window (30, 90, or 180 days)
114    pub window_days: Option<u32>,
115}
116
117/// Response for forecast list
118#[derive(Debug, Serialize)]
119pub struct ForecastListResponse {
120    /// Forecasts
121    pub forecasts: Vec<ChangeForecast>,
122    /// Total count
123    pub total: usize,
124}
125
126/// Get forecasts
127///
128/// GET /api/v1/forecasts
129#[cfg(feature = "database")]
130pub async fn list_forecasts(
131    State(state): State<ForecastingState>,
132    Query(params): Query<ListForecastsQuery>,
133) -> Result<Json<ForecastListResponse>, StatusCode> {
134    let pool = match state.database.as_ref().and_then(|db| db.pool()) {
135        Some(pool) => pool,
136        None => {
137            return Ok(Json(ForecastListResponse {
138                forecasts: Vec::new(),
139                total: 0,
140            }));
141        }
142    };
143
144    // Build query with filters
145    let mut query = String::from(
146        "SELECT id, workspace_id, service_id, service_name, endpoint, method,
147         forecast_window_days, predicted_change_probability, predicted_break_probability,
148         next_expected_change_date, next_expected_break_date, volatility_score, confidence,
149         seasonal_patterns, predicted_at, expires_at
150         FROM api_change_forecasts WHERE expires_at > NOW()",
151    );
152
153    let mut bind_index = 1;
154
155    if params.workspace_id.is_some() {
156        query.push_str(&format!(" AND workspace_id = ${}", bind_index));
157        bind_index += 1;
158    }
159
160    if params.service_id.is_some() {
161        query.push_str(&format!(" AND service_id = ${}", bind_index));
162        bind_index += 1;
163    }
164
165    if params.endpoint.is_some() {
166        query.push_str(&format!(" AND endpoint = ${}", bind_index));
167        bind_index += 1;
168    }
169
170    if params.method.is_some() {
171        query.push_str(&format!(" AND method = ${}", bind_index));
172        bind_index += 1;
173    }
174
175    if let Some(_window) = params.window_days {
176        query.push_str(&format!(" AND forecast_window_days = ${}", bind_index));
177    }
178
179    query.push_str(" ORDER BY predicted_at DESC LIMIT 100");
180
181    // Build query with proper bindings using sqlx
182    let mut query_builder = sqlx::query(&query);
183
184    if let Some(ws_id) = &params.workspace_id {
185        let uuid = Uuid::parse_str(ws_id).ok();
186        query_builder = query_builder.bind(uuid);
187    }
188
189    if let Some(svc_id) = &params.service_id {
190        query_builder = query_builder.bind(svc_id);
191    }
192
193    if let Some(ep) = &params.endpoint {
194        query_builder = query_builder.bind(ep);
195    }
196
197    if let Some(m) = &params.method {
198        query_builder = query_builder.bind(m);
199    }
200
201    if let Some(window) = params.window_days {
202        query_builder = query_builder.bind(window as i32);
203    }
204
205    // Execute query
206    let rows = query_builder.fetch_all(pool).await.map_err(|e| {
207        tracing::error!("Failed to query forecasts: {}", e);
208        StatusCode::INTERNAL_SERVER_ERROR
209    })?;
210
211    // Map rows to ChangeForecast
212    let mut forecasts = Vec::new();
213    for row in rows {
214        match map_row_to_change_forecast(&row) {
215            Ok(forecast) => forecasts.push(forecast),
216            Err(e) => {
217                tracing::warn!("Failed to map forecast row: {}", e);
218                continue;
219            }
220        }
221    }
222
223    let total = forecasts.len();
224    Ok(Json(ForecastListResponse { forecasts, total }))
225}
226
227/// List forecasts (no database)
228///
229/// GET /api/v1/forecasts
230#[cfg(not(feature = "database"))]
231pub async fn list_forecasts(
232    State(_state): State<ForecastingState>,
233    Query(_params): Query<ListForecastsQuery>,
234) -> Result<Json<ForecastListResponse>, StatusCode> {
235    Ok(Json(ForecastListResponse {
236        forecasts: Vec::new(),
237        total: 0,
238    }))
239}
240
241/// Get service-level forecasts
242///
243/// GET /api/v1/forecasts/service/{service_id}
244#[cfg(feature = "database")]
245pub async fn get_service_forecasts(
246    State(state): State<ForecastingState>,
247    Path(service_id): Path<String>,
248    Query(_params): Query<ListForecastsQuery>,
249) -> Result<Json<ForecastListResponse>, StatusCode> {
250    let pool = match state.database.as_ref().and_then(|db| db.pool()) {
251        Some(pool) => pool,
252        None => {
253            return Ok(Json(ForecastListResponse {
254                forecasts: Vec::new(),
255                total: 0,
256            }));
257        }
258    };
259
260    // Query forecasts for this service
261    let rows = sqlx::query(
262        "SELECT * FROM api_change_forecasts
263         WHERE service_id = $1 AND expires_at > NOW()
264         ORDER BY predicted_at DESC LIMIT 50",
265    )
266    .bind(&service_id)
267    .fetch_all(pool)
268    .await
269    .map_err(|e| {
270        tracing::error!("Failed to query service forecasts: {}", e);
271        StatusCode::INTERNAL_SERVER_ERROR
272    })?;
273
274    // Map rows to forecasts
275    let mut forecasts = Vec::new();
276    for row in rows {
277        match map_row_to_change_forecast(&row) {
278            Ok(forecast) => forecasts.push(forecast),
279            Err(e) => {
280                tracing::warn!("Failed to map service forecast row: {}", e);
281                continue;
282            }
283        }
284    }
285
286    let total = forecasts.len();
287    Ok(Json(ForecastListResponse { forecasts, total }))
288}
289
290/// Get service-level forecasts (no database)
291///
292/// GET /api/v1/forecasts/service/{service_id}
293#[cfg(not(feature = "database"))]
294pub async fn get_service_forecasts(
295    State(_state): State<ForecastingState>,
296    Path(_service_id): Path<String>,
297    Query(_params): Query<ListForecastsQuery>,
298) -> Result<Json<ForecastListResponse>, StatusCode> {
299    Ok(Json(ForecastListResponse {
300        forecasts: Vec::new(),
301        total: 0,
302    }))
303}
304
305/// Get endpoint-level forecasts
306///
307/// GET /api/v1/forecasts/endpoint/{endpoint}
308#[cfg(feature = "database")]
309pub async fn get_endpoint_forecasts(
310    State(state): State<ForecastingState>,
311    Path(endpoint): Path<String>,
312    Query(params): Query<ListForecastsQuery>,
313) -> Result<Json<ForecastListResponse>, StatusCode> {
314    let pool = match state.database.as_ref().and_then(|db| db.pool()) {
315        Some(pool) => pool,
316        None => {
317            return Ok(Json(ForecastListResponse {
318                forecasts: Vec::new(),
319                total: 0,
320            }));
321        }
322    };
323
324    let method = params.method.as_deref().unwrap_or("%");
325
326    let rows = sqlx::query(
327        "SELECT * FROM api_change_forecasts
328         WHERE endpoint = $1 AND method LIKE $2 AND expires_at > NOW()
329         ORDER BY predicted_at DESC LIMIT 50",
330    )
331    .bind(&endpoint)
332    .bind(method)
333    .fetch_all(pool)
334    .await
335    .map_err(|e| {
336        tracing::error!("Failed to query endpoint forecasts: {}", e);
337        StatusCode::INTERNAL_SERVER_ERROR
338    })?;
339
340    // Map rows to forecasts
341    let mut forecasts = Vec::new();
342    for row in rows {
343        match map_row_to_change_forecast(&row) {
344            Ok(forecast) => forecasts.push(forecast),
345            Err(e) => {
346                tracing::warn!("Failed to map endpoint forecast row: {}", e);
347                continue;
348            }
349        }
350    }
351
352    let total = forecasts.len();
353    Ok(Json(ForecastListResponse { forecasts, total }))
354}
355
356/// Get endpoint-level forecasts (no database)
357///
358/// GET /api/v1/forecasts/endpoint/{endpoint}
359#[cfg(not(feature = "database"))]
360pub async fn get_endpoint_forecasts(
361    State(_state): State<ForecastingState>,
362    Path(_endpoint): Path<String>,
363    Query(_params): Query<ListForecastsQuery>,
364) -> Result<Json<ForecastListResponse>, StatusCode> {
365    Ok(Json(ForecastListResponse {
366        forecasts: Vec::new(),
367        total: 0,
368    }))
369}
370
371/// Request to refresh forecasts
372#[derive(Debug, Deserialize)]
373pub struct RefreshForecastsRequest {
374    /// Workspace ID
375    pub workspace_id: Option<String>,
376    /// Service ID
377    pub service_id: Option<String>,
378    /// Endpoint (optional)
379    pub endpoint: Option<String>,
380    /// Method (optional)
381    pub method: Option<String>,
382}
383
384/// Refresh forecasts
385///
386/// POST /api/v1/forecasts/refresh
387#[cfg(feature = "database")]
388pub async fn refresh_forecasts(
389    State(state): State<ForecastingState>,
390    Json(request): Json<RefreshForecastsRequest>,
391) -> Result<Json<serde_json::Value>, StatusCode> {
392    let pool = match state.database.as_ref().and_then(|db| db.pool()) {
393        Some(pool) => pool,
394        None => {
395            return Ok(Json(serde_json::json!({
396                "success": false,
397                "error": "Database not available"
398            })));
399        }
400    };
401
402    // Query historical incidents for forecasting
403    let mut incident_query = String::from(
404        "SELECT id, workspace_id, endpoint, method, incident_type, severity, status,
405         detected_at, details, created_at, updated_at
406         FROM drift_incidents WHERE 1=1",
407    );
408
409    if let Some(_ws_id) = &request.workspace_id {
410        incident_query.push_str(" AND workspace_id = $1");
411    }
412
413    // Execute query to get incidents
414    let rows = sqlx::query(&incident_query).fetch_all(pool).await.map_err(|e| {
415        tracing::error!("Failed to query drift incidents: {}", e);
416        StatusCode::INTERNAL_SERVER_ERROR
417    })?;
418
419    // Map rows to DriftIncident and generate forecasts
420    use mockforge_foundation::incidents_types::{IncidentSeverity, IncidentStatus, IncidentType};
421    use sqlx::Row;
422    let mut incidents = Vec::new();
423    for row in rows {
424        let id: Uuid = row.try_get("id").map_err(|e| {
425            tracing::error!("Failed to get id from row: {}", e);
426            StatusCode::INTERNAL_SERVER_ERROR
427        })?;
428        let workspace_id: Option<Uuid> = row.try_get("workspace_id").ok();
429        let endpoint: String = match row.try_get("endpoint") {
430            Ok(e) => e,
431            Err(_) => continue,
432        };
433        let method: String = match row.try_get("method") {
434            Ok(m) => m,
435            Err(_) => continue,
436        };
437        let incident_type_str: String = match row.try_get("incident_type") {
438            Ok(s) => s,
439            Err(_) => continue,
440        };
441        let severity_str: String = match row.try_get("severity") {
442            Ok(s) => s,
443            Err(_) => continue,
444        };
445        let status_str: String = match row.try_get("status") {
446            Ok(s) => s,
447            Err(_) => continue,
448        };
449        let detected_at: DateTime<Utc> = match row.try_get("detected_at") {
450            Ok(dt) => dt,
451            Err(_) => continue,
452        };
453        let details_json: serde_json::Value = row.try_get("details").unwrap_or_default();
454        let created_at: DateTime<Utc> = match row.try_get("created_at") {
455            Ok(dt) => dt,
456            Err(_) => continue,
457        };
458        let updated_at: DateTime<Utc> = match row.try_get("updated_at") {
459            Ok(dt) => dt,
460            Err(_) => continue,
461        };
462
463        let incident_type = match incident_type_str.as_str() {
464            "breaking_change" => IncidentType::BreakingChange,
465            "threshold_exceeded" => IncidentType::ThresholdExceeded,
466            _ => continue, // Skip invalid types
467        };
468
469        let severity = match severity_str.as_str() {
470            "low" => IncidentSeverity::Low,
471            "medium" => IncidentSeverity::Medium,
472            "high" => IncidentSeverity::High,
473            "critical" => IncidentSeverity::Critical,
474            _ => continue, // Skip invalid severity
475        };
476
477        let status = match status_str.as_str() {
478            "open" => IncidentStatus::Open,
479            "acknowledged" => IncidentStatus::Acknowledged,
480            "resolved" => IncidentStatus::Resolved,
481            "closed" => IncidentStatus::Closed,
482            _ => continue, // Skip invalid status
483        };
484
485        incidents.push(DriftIncident {
486            id: id.to_string(),
487            budget_id: None,
488            workspace_id: workspace_id.map(|u| u.to_string()),
489            endpoint,
490            method,
491            incident_type,
492            severity,
493            status,
494            detected_at: detected_at.timestamp(),
495            resolved_at: None,
496            details: details_json,
497            external_ticket_id: None,
498            external_ticket_url: None,
499            created_at: created_at.timestamp(),
500            updated_at: updated_at.timestamp(),
501            sync_cycle_id: None,
502            contract_diff_id: None,
503            before_sample: None,
504            after_sample: None,
505            fitness_test_results: Vec::new(),
506            affected_consumers: None,
507            protocol: None,
508        });
509    }
510
511    // Generate forecasts from incidents by grouping by endpoint/method
512    use mockforge_foundation::incidents_types::DriftIncident;
513    use std::collections::HashMap;
514    let mut forecasts_generated = 0;
515    let mut endpoint_groups: HashMap<(String, String), Vec<DriftIncident>> = HashMap::new();
516
517    for incident in incidents {
518        endpoint_groups
519            .entry((incident.endpoint.clone(), incident.method.clone()))
520            .or_default()
521            .push(incident);
522    }
523
524    for ((endpoint, method), group_incidents) in endpoint_groups {
525        if let Some(forecast) = state.forecaster.generate_forecast(
526            &group_incidents,
527            request.workspace_id.clone(),
528            None, // service_id
529            None, // service_name
530            endpoint,
531            method,
532            30, // forecast_window_days
533        ) {
534            // Persist forecast to database
535            if let Err(e) = store_forecast(pool, &forecast, request.workspace_id.as_deref()).await {
536                tracing::warn!("Failed to store forecast: {}", e);
537            }
538            forecasts_generated += 1;
539        }
540    }
541
542    Ok(Json(serde_json::json!({
543        "success": true,
544        "message": "Forecasts refreshed",
545        "forecasts_generated": forecasts_generated
546    })))
547}
548
549/// Refresh forecasts (no database)
550///
551/// POST /api/v1/forecasts/refresh
552#[cfg(not(feature = "database"))]
553pub async fn refresh_forecasts(
554    State(_state): State<ForecastingState>,
555    Json(_request): Json<RefreshForecastsRequest>,
556) -> Result<Json<serde_json::Value>, StatusCode> {
557    Ok(Json(serde_json::json!({
558        "success": false,
559        "error": "Database not available"
560    })))
561}
562
563/// Store a forecast in the database
564#[cfg(feature = "database")]
565pub async fn store_forecast(
566    pool: &sqlx::PgPool,
567    forecast: &ChangeForecast,
568    workspace_id: Option<&str>,
569) -> Result<(), sqlx::Error> {
570    let id = Uuid::new_v4();
571    let workspace_uuid = workspace_id.and_then(|id| Uuid::parse_str(id).ok());
572
573    sqlx::query(
574        r#"
575        INSERT INTO api_change_forecasts (
576            id, workspace_id, service_id, service_name, endpoint, method,
577            forecast_window_days, predicted_change_probability, predicted_break_probability,
578            next_expected_change_date, next_expected_break_date, volatility_score, confidence,
579            seasonal_patterns, predicted_at, expires_at
580        ) VALUES (
581            $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16
582        )
583        ON CONFLICT (workspace_id, service_id, endpoint, method, forecast_window_days)
584        DO UPDATE SET
585            predicted_change_probability = EXCLUDED.predicted_change_probability,
586            predicted_break_probability = EXCLUDED.predicted_break_probability,
587            next_expected_change_date = EXCLUDED.next_expected_change_date,
588            next_expected_break_date = EXCLUDED.next_expected_break_date,
589            volatility_score = EXCLUDED.volatility_score,
590            confidence = EXCLUDED.confidence,
591            seasonal_patterns = EXCLUDED.seasonal_patterns,
592            predicted_at = EXCLUDED.predicted_at,
593            expires_at = EXCLUDED.expires_at,
594            updated_at = NOW()
595        "#,
596    )
597    .bind(id)
598    .bind(workspace_uuid)
599    .bind(forecast.service_id.as_deref())
600    .bind(forecast.service_name.as_deref())
601    .bind(&forecast.endpoint)
602    .bind(&forecast.method)
603    .bind(forecast.forecast_window_days as i32)
604    .bind(forecast.predicted_change_probability)
605    .bind(forecast.predicted_break_probability)
606    .bind(forecast.next_expected_change_date)
607    .bind(forecast.next_expected_break_date)
608    .bind(forecast.volatility_score)
609    .bind(forecast.confidence)
610    .bind(serde_json::to_value(&forecast.seasonal_patterns).unwrap_or_default())
611    .bind(forecast.predicted_at)
612    .bind(forecast.expires_at)
613    .execute(pool)
614    .await?;
615
616    Ok(())
617}
618
619/// Create router for forecasting endpoints
620pub fn forecasting_router(state: ForecastingState) -> axum::Router {
621    use axum::routing::{get, post};
622    use axum::Router;
623
624    Router::new()
625        .route("/api/v1/forecasts", get(list_forecasts))
626        .route("/api/v1/forecasts/service/{service_id}", get(get_service_forecasts))
627        .route("/api/v1/forecasts/endpoint/{endpoint}", get(get_endpoint_forecasts))
628        .route("/api/v1/forecasts/refresh", post(refresh_forecasts))
629        .with_state(state)
630}
mockforge_intelligence/handlers/forecasting.rs

mockforge_intelligence/handlers/
forecasting.rs
