mockforge_http/middleware/
drift_tracking.rs

1//! Drift tracking middleware
2//!
3//! This middleware integrates drift budget evaluation and consumer usage tracking
4//! with contract diff analysis.
5
6// Uses DriftBudgetEngine + ContractDiffAnalyzer which stay in core.
7#![allow(deprecated)]
8
9use axum::{body::Body, extract::Request, http::Response, middleware::Next};
10use mockforge_contracts::consumer_contracts::{ConsumerBreakingChangeDetector, UsageRecorder};
11use mockforge_core::{
12    ai_contract_diff::ContractDiffAnalyzer,
13    contract_drift::DriftBudgetEngine,
14    incidents::{IncidentManager, IncidentSeverity, IncidentType},
15    openapi::OpenApiSpec,
16};
17use mockforge_foundation::contract_drift_types::DriftResult;
18use serde_json::Value;
19use std::collections::HashMap;
20use std::sync::Arc;
21use tracing::{debug, warn};
22
23/// Maximum request body size to buffer for drift tracking.
24///
25/// Issue #79 — see `contract_diff_middleware`. Same root cause: the
26/// old hard-coded 1 MiB cap, paired with the over-cap branch
27/// substituting `Body::empty()`, broke downstream handlers on large
28/// chunked uploads. Configurable via `MOCKFORGE_DRIFT_MAX_BODY_MB`.
29fn max_drift_body_size() -> usize {
30    const DEFAULT_MB: usize = 10;
31    std::env::var("MOCKFORGE_DRIFT_MAX_BODY_MB")
32        .ok()
33        .and_then(|v| v.parse::<usize>().ok())
34        .unwrap_or(DEFAULT_MB)
35        .saturating_mul(1024 * 1024)
36}
37
38/// State for drift tracking middleware
39#[derive(Clone)]
40pub struct DriftTrackingState {
41    /// Contract diff analyzer
42    pub diff_analyzer: Option<Arc<ContractDiffAnalyzer>>,
43    /// OpenAPI spec (if available)
44    pub spec: Option<Arc<OpenApiSpec>>,
45    /// Drift budget engine
46    pub drift_engine: Arc<DriftBudgetEngine>,
47    /// Incident manager
48    pub incident_manager: Arc<IncidentManager>,
49    /// Usage recorder for consumer contracts
50    pub usage_recorder: Arc<UsageRecorder>,
51    /// Consumer breaking change detector
52    pub consumer_detector: Arc<ConsumerBreakingChangeDetector>,
53    /// Whether drift tracking is enabled
54    pub enabled: bool,
55}
56
57/// Middleware to track drift and consumer usage (with state from extensions)
58///
59/// This middleware requires response body buffering middleware to be applied first.
60/// The response body is extracted from buffered response extensions.
61pub async fn drift_tracking_middleware_with_extensions(
62    req: Request<Body>,
63    next: Next,
64) -> Response<Body> {
65    // Extract state from request extensions
66    let state = req.extensions().get::<DriftTrackingState>().cloned();
67
68    let state = if let Some(state) = state {
69        state
70    } else {
71        // No state available, skip drift tracking
72        return next.run(req).await;
73    };
74
75    if !state.enabled {
76        return next.run(req).await;
77    }
78
79    let method = req.method().to_string();
80    let path = req.uri().path().to_string();
81    let max_body = max_drift_body_size();
82
83    // Issue #79 — pre-check Content-Length so we skip capture cleanly
84    // for over-cap requests instead of consuming the body and then
85    // substituting `Body::empty()` (which broke downstream handlers).
86    let content_length = req
87        .headers()
88        .get(http::header::CONTENT_LENGTH)
89        .and_then(|v| v.to_str().ok())
90        .and_then(|s| s.parse::<usize>().ok());
91    if let Some(len) = content_length {
92        if len > max_body {
93            debug!(
94                "drift_tracking: skipping capture for {} {} — content-length {} > cap {}",
95                method, path, len, max_body
96            );
97            return next.run(req).await;
98        }
99    }
100
101    // Extract consumer identifier and headers from request
102    let consumer_id = extract_consumer_id(&req);
103
104    // Extract headers for capture
105    let captured_headers = extract_headers_for_capture(&req);
106
107    // Buffer the request body so we can capture it and still forward it
108    let (parts, body) = req.into_parts();
109    let body_bytes = match axum::body::to_bytes(body, max_body).await {
110        Ok(b) => b,
111        Err(_) => {
112            // Chunked over-cap body — body partially consumed and we
113            // cannot rebuild. 413 PayloadTooLarge instead of the old
114            // silent `Body::empty()` substitution. Issue #79.
115            return Response::builder()
116                .status(http::StatusCode::PAYLOAD_TOO_LARGE)
117                .header(
118                    http::header::CONTENT_TYPE,
119                    "application/json",
120                )
121                .body(Body::from(format!(
122                    r#"{{"error":"PAYLOAD_TOO_LARGE","message":"chunked request body exceeded drift_tracking capture cap (~{} MiB); raise MOCKFORGE_DRIFT_MAX_BODY_MB or send Content-Length"}}"#,
123                    max_body / (1024 * 1024)
124                )))
125                .unwrap_or_else(|_| Response::new(Body::from("payload too large")));
126        }
127    };
128
129    // Try to parse body as JSON for structured capture
130    let captured_body = if !body_bytes.is_empty() {
131        serde_json::from_slice::<Value>(&body_bytes).ok()
132    } else {
133        None
134    };
135
136    // Reconstruct the request with the buffered body
137    let req = Request::from_parts(parts, Body::from(body_bytes));
138
139    // Process request and get response
140    let response = next.run(req).await;
141
142    // Extract response body for consumer usage tracking
143    let response_body = extract_response_body(&response);
144
145    // Record consumer usage if consumer is identified
146    if let Some(ref consumer_id) = consumer_id {
147        if let Some(body) = &response_body {
148            state.usage_recorder.record_usage(consumer_id, &path, &method, Some(body)).await;
149        }
150    }
151
152    // Perform contract diff analysis if analyzer and spec are available
153    if let (Some(ref analyzer), Some(ref spec)) = (&state.diff_analyzer, &state.spec) {
154        // Create captured request with body and headers
155        let mut captured = mockforge_core::ai_contract_diff::CapturedRequest::new(
156            &method,
157            &path,
158            "drift_tracking",
159        )
160        .with_headers(captured_headers)
161        .with_response(response.status().as_u16(), response_body.clone());
162
163        if let Some(body_value) = captured_body {
164            captured = captured.with_body(body_value);
165        }
166
167        // Analyze request against contract
168        match analyzer.analyze(&captured, spec).await {
169            Ok(diff_result) => {
170                // Evaluate drift budget
171                let drift_result = state.drift_engine.evaluate(&diff_result, &path, &method);
172
173                // Emit Prometheus gauges for the drift dashboard (#678). The
174                // workspace_id is not yet plumbed through the middleware —
175                // pass an empty label so a single global series records,
176                // matching the pattern used for the non-workspace
177                // `pillar_tracking` calls below. When workspace identification
178                // lands we can switch this label to the resolved id.
179                mockforge_observability::get_global_registry().record_drift_evaluation(
180                    mockforge_observability::DriftEvaluationSample {
181                        workspace_id: "",
182                        endpoint: &path,
183                        method: &method,
184                        total: drift_result.metrics.total_changes,
185                        breaking: drift_result.breaking_changes,
186                        potentially_breaking: drift_result.potentially_breaking_changes,
187                        budget_exceeded: drift_result.budget_exceeded,
188                    },
189                );
190
191                // Record contracts pillar usage for drift detection
192                mockforge_core::pillar_tracking::record_contracts_usage(
193                    None, // workspace_id could be extracted from request if available
194                    None,
195                    "drift_detection",
196                    serde_json::json!({
197                        "endpoint": path,
198                        "method": method,
199                        "breaking_changes": drift_result.breaking_changes,
200                        "non_breaking_changes": drift_result.non_breaking_changes,
201                        "incident": drift_result.should_create_incident
202                    }),
203                )
204                .await;
205
206                // If this endpoint has a drift budget configured, record that so
207                // the Contracts pillar dashboard can count distinct budgeted endpoints.
208                let endpoint_key = format!("{} {}", method, path);
209                let budget_config = state.drift_engine.config();
210                if budget_config.enabled
211                    && (budget_config.per_endpoint_budgets.contains_key(&endpoint_key)
212                        || budget_config.default_budget.is_some())
213                {
214                    mockforge_core::pillar_tracking::record_contracts_usage(
215                        None,
216                        None,
217                        "drift_budget_configured",
218                        serde_json::json!({
219                            "endpoint": endpoint_key,
220                        }),
221                    )
222                    .await;
223                }
224
225                // Create incident if budget is exceeded or breaking changes detected
226                if drift_result.should_create_incident {
227                    let incident_type = if drift_result.breaking_changes > 0 {
228                        IncidentType::BreakingChange
229                    } else {
230                        IncidentType::ThresholdExceeded
231                    };
232
233                    let severity = determine_severity(&drift_result);
234
235                    let details = serde_json::json!({
236                        "breaking_changes": drift_result.breaking_changes,
237                        "non_breaking_changes": drift_result.non_breaking_changes,
238                        "breaking_mismatches": drift_result.breaking_mismatches,
239                        "non_breaking_mismatches": drift_result.non_breaking_mismatches,
240                        "budget_exceeded": drift_result.budget_exceeded,
241                    });
242
243                    // Create incident with before/after samples
244                    // Extract before/after samples from diff result if available
245                    let before_sample = Some(serde_json::json!({
246                        "contract_format": diff_result.metadata.contract_format,
247                        "contract_version": diff_result.metadata.contract_version,
248                        "endpoint": path,
249                        "method": method,
250                    }));
251
252                    let after_sample = Some(serde_json::json!({
253                        "mismatches": diff_result.mismatches,
254                        "recommendations": diff_result.recommendations,
255                        "corrections": diff_result.corrections,
256                    }));
257
258                    let _incident = state
259                        .incident_manager
260                        .create_incident_with_samples(
261                            path.clone(),
262                            method.clone(),
263                            incident_type,
264                            severity,
265                            details,
266                            None, // budget_id
267                            None, // workspace_id
268                            None, // sync_cycle_id
269                            None, // contract_diff_id (could be generated from diff_result)
270                            before_sample,
271                            after_sample,
272                            Some(drift_result.fitness_test_results.clone()), // fitness_test_results
273                            drift_result.consumer_impact.clone(),            // affected_consumers
274                            Some(mockforge_foundation::protocol::Protocol::Http), // protocol
275                        )
276                        .await;
277
278                    warn!(
279                        "Drift incident created: {} {} - {} breaking changes, {} non-breaking changes",
280                        method, path, drift_result.breaking_changes, drift_result.non_breaking_changes
281                    );
282                }
283
284                // Check for consumer-specific violations
285                if let Some(ref consumer_id) = consumer_id {
286                    let violations = state
287                        .consumer_detector
288                        .detect_violations(consumer_id, &path, &method, &diff_result, None)
289                        .await;
290
291                    if !violations.is_empty() {
292                        warn!(
293                            "Consumer {} has {} violations on {} {}",
294                            consumer_id,
295                            violations.len(),
296                            method,
297                            path
298                        );
299                    }
300                }
301            }
302            Err(e) => {
303                debug!("Contract diff analysis failed: {}", e);
304            }
305        }
306    }
307
308    response
309}
310
311/// Extract consumer identifier from request
312fn extract_consumer_id(req: &Request<Body>) -> Option<String> {
313    // Try to extract from various sources:
314    // 1. X-Consumer-ID header
315    if let Some(consumer_id) = req.headers().get("x-consumer-id").and_then(|h| h.to_str().ok()) {
316        return Some(consumer_id.to_string());
317    }
318
319    // 2. X-Workspace-ID header (for workspace-based consumers)
320    if let Some(workspace_id) = req.headers().get("x-workspace-id").and_then(|h| h.to_str().ok()) {
321        return Some(format!("workspace:{}", workspace_id));
322    }
323
324    // 3. API key from header
325    if let Some(api_key) = req
326        .headers()
327        .get("x-api-key")
328        .or_else(|| req.headers().get("authorization"))
329        .and_then(|h| h.to_str().ok())
330    {
331        // Hash the API key for privacy
332        use sha2::{Digest, Sha256};
333        let mut hasher = Sha256::new();
334        hasher.update(api_key.as_bytes());
335        let hash = format!("{:x}", hasher.finalize());
336        return Some(format!("api_key:{}", hash));
337    }
338
339    None
340}
341
342/// Extract safe headers for drift capture
343fn extract_headers_for_capture(req: &Request<Body>) -> HashMap<String, String> {
344    let safe_headers = [
345        "accept",
346        "accept-encoding",
347        "content-type",
348        "content-length",
349        "user-agent",
350    ];
351    let mut captured = HashMap::new();
352    for name in safe_headers {
353        if let Some(value) = req.headers().get(name).and_then(|v| v.to_str().ok()) {
354            captured.insert(name.to_string(), value.to_string());
355        }
356    }
357    captured
358}
359
360/// Extract response body as JSON value
361fn extract_response_body(response: &Response<Body>) -> Option<Value> {
362    // Try to get buffered response from extensions
363    if let Some(buffered) = crate::middleware::get_buffered_response(response) {
364        return buffered.json();
365    }
366
367    // If not buffered, try to parse from response body
368    // Note: This requires the response body to be buffered by upstream middleware
369    None
370}
371
372/// Determine incident severity from drift result
373fn determine_severity(drift_result: &DriftResult) -> IncidentSeverity {
374    if drift_result.breaking_changes > 0 {
375        // Check if any breaking mismatch is critical
376        if drift_result
377            .breaking_mismatches
378            .iter()
379            .any(|m| m.severity == mockforge_core::ai_contract_diff::MismatchSeverity::Critical)
380        {
381            return IncidentSeverity::Critical;
382        }
383        // Check if any breaking mismatch is high
384        if drift_result
385            .breaking_mismatches
386            .iter()
387            .any(|m| m.severity == mockforge_core::ai_contract_diff::MismatchSeverity::High)
388        {
389            return IncidentSeverity::High;
390        }
391        return IncidentSeverity::Medium;
392    }
393
394    // Non-breaking changes are lower severity
395    if drift_result.non_breaking_changes > 5 {
396        IncidentSeverity::Medium
397    } else {
398        IncidentSeverity::Low
399    }
400}
mockforge_http/middleware/drift_tracking.rs

mockforge_http/middleware/
drift_tracking.rs
