1use chrono::{DateTime, Utc};
4use serde::{Deserialize, Serialize};
5use std::collections::{HashMap, VecDeque};
6use std::sync::Arc;
7use tokio::sync::RwLock;
8
9#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
13#[serde(rename_all = "lowercase")]
14pub enum RealityContinuumType {
15 Synthetic,
17 Blended,
19 Live,
21}
22
23impl RealityContinuumType {
24 pub fn from_blend_ratio(ratio: f64) -> Self {
26 if ratio <= 0.0 {
27 Self::Synthetic
28 } else if ratio >= 1.0 {
29 Self::Live
30 } else {
31 Self::Blended
32 }
33 }
34
35 pub fn name(&self) -> &'static str {
37 match self {
38 RealityContinuumType::Synthetic => "Synthetic",
39 RealityContinuumType::Blended => "Blended",
40 RealityContinuumType::Live => "Live",
41 }
42 }
43}
44
45#[derive(Debug, Clone, Serialize, Deserialize)]
47pub struct DataSourceBreakdown {
48 #[serde(default)]
50 pub recorded_percent: f64,
51 #[serde(default)]
53 pub generator_percent: f64,
54 #[serde(default)]
56 pub upstream_percent: f64,
57}
58
59impl Default for DataSourceBreakdown {
60 fn default() -> Self {
61 Self {
62 recorded_percent: 0.0,
63 generator_percent: 100.0,
64 upstream_percent: 0.0,
65 }
66 }
67}
68
69impl DataSourceBreakdown {
70 pub fn from_blend_ratio(blend_ratio: f64, recorded_ratio: f64) -> Self {
75 let upstream = blend_ratio * (1.0 - recorded_ratio);
76 let generator = (1.0 - blend_ratio) * (1.0 - recorded_ratio);
77 let recorded = recorded_ratio;
78
79 Self {
80 recorded_percent: recorded * 100.0,
81 generator_percent: generator * 100.0,
82 upstream_percent: upstream * 100.0,
83 }
84 }
85
86 pub fn normalize(&mut self) {
88 let total = self.recorded_percent + self.generator_percent + self.upstream_percent;
89 if total > 0.0 {
90 self.recorded_percent = (self.recorded_percent / total) * 100.0;
91 self.generator_percent = (self.generator_percent / total) * 100.0;
92 self.upstream_percent = (self.upstream_percent / total) * 100.0;
93 }
94 }
95}
96
97#[derive(Debug, Clone, Serialize, Deserialize)]
102pub struct RealityTraceMetadata {
103 #[serde(skip_serializing_if = "Option::is_none")]
105 pub reality_level: Option<crate::reality::RealityLevel>,
106 pub reality_continuum_type: RealityContinuumType,
108 #[serde(default)]
110 pub blend_ratio: f64,
111 pub data_source_breakdown: DataSourceBreakdown,
113 #[serde(skip_serializing_if = "Option::is_none")]
115 pub active_persona_id: Option<String>,
116 #[serde(skip_serializing_if = "Option::is_none")]
118 pub active_scenario: Option<String>,
119 #[serde(default)]
121 pub active_chaos_profiles: Vec<String>,
122 #[serde(default)]
124 pub active_latency_profiles: Vec<String>,
125}
126
127impl Default for RealityTraceMetadata {
128 fn default() -> Self {
129 Self {
130 reality_level: None,
131 reality_continuum_type: RealityContinuumType::Synthetic,
132 blend_ratio: 0.0,
133 data_source_breakdown: DataSourceBreakdown::default(),
134 active_persona_id: None,
135 active_scenario: None,
136 active_chaos_profiles: Vec::new(),
137 active_latency_profiles: Vec::new(),
138 }
139 }
140}
141
142impl RealityTraceMetadata {
143 pub fn from_unified_state(
148 unified_state: &crate::consistency::types::UnifiedState,
149 blend_ratio: f64,
150 path: &str,
151 ) -> Self {
152 let reality_continuum_type = RealityContinuumType::from_blend_ratio(blend_ratio);
153
154 let active_chaos_profiles: Vec<String> = unified_state
156 .active_chaos_rules
157 .iter()
158 .filter_map(|r| r.get("name").and_then(|v| v.as_str()).map(|s| s.to_string()))
159 .collect();
160
161 let active_latency_profiles = Vec::new();
164
165 let mut breakdown = DataSourceBreakdown::from_blend_ratio(blend_ratio, 0.0);
168 breakdown.normalize();
169
170 Self {
171 reality_level: Some(unified_state.reality_level),
172 reality_continuum_type,
173 blend_ratio,
174 data_source_breakdown: breakdown,
175 active_persona_id: unified_state.active_persona.as_ref().map(|p| p.id.clone()),
176 active_scenario: unified_state.active_scenario.clone(),
177 active_chaos_profiles,
178 active_latency_profiles,
179 }
180 }
181}
182
183#[derive(Debug, Clone, Serialize, Deserialize)]
185pub struct RequestLogEntry {
186 pub id: String,
188 pub timestamp: DateTime<Utc>,
190 pub server_type: String,
192 pub method: String,
194 pub path: String,
196 pub status_code: u16,
198 pub response_time_ms: u64,
200 pub client_ip: Option<String>,
202 pub user_agent: Option<String>,
204 pub headers: HashMap<String, String>,
206 #[serde(default, skip_serializing_if = "HashMap::is_empty")]
208 pub query_params: HashMap<String, String>,
209 pub response_size_bytes: u64,
211 pub error_message: Option<String>,
213 pub metadata: HashMap<String, String>,
215 #[serde(skip_serializing_if = "Option::is_none")]
220 pub reality_metadata: Option<RealityTraceMetadata>,
221}
222
223#[derive(Debug, Clone)]
225pub struct CentralizedRequestLogger {
226 logs: Arc<RwLock<VecDeque<RequestLogEntry>>>,
228 max_logs: usize,
230}
231
232impl Default for CentralizedRequestLogger {
233 fn default() -> Self {
234 Self::new(1000) }
236}
237
238impl CentralizedRequestLogger {
239 pub fn new(max_logs: usize) -> Self {
241 Self {
242 logs: Arc::new(RwLock::new(VecDeque::new())),
243 max_logs,
244 }
245 }
246
247 pub async fn log_request(&self, entry: RequestLogEntry) {
249 let mut logs = self.logs.write().await;
250
251 logs.push_front(entry);
253
254 while logs.len() > self.max_logs {
256 logs.pop_back();
257 }
258 }
259
260 pub async fn get_recent_logs(&self, limit: Option<usize>) -> Vec<RequestLogEntry> {
262 let logs = self.logs.read().await;
263 let take_count = limit.unwrap_or(logs.len()).min(logs.len());
264 logs.iter().take(take_count).cloned().collect()
265 }
266
267 pub async fn get_logs_by_server(
269 &self,
270 server_type: &str,
271 limit: Option<usize>,
272 ) -> Vec<RequestLogEntry> {
273 let logs = self.logs.read().await;
274 logs.iter()
275 .filter(|log| log.server_type == server_type)
276 .take(limit.unwrap_or(logs.len()))
277 .cloned()
278 .collect()
279 }
280
281 pub async fn get_request_counts_by_server(&self) -> HashMap<String, u64> {
283 let logs = self.logs.read().await;
284 let mut counts = HashMap::new();
285
286 for log in logs.iter() {
287 *counts.entry(log.server_type.clone()).or_insert(0) += 1;
288 }
289
290 counts
291 }
292
293 pub async fn clear_logs(&self) {
295 let mut logs = self.logs.write().await;
296 logs.clear();
297 }
298
299 pub async fn find_matching_requests(
305 &self,
306 pattern: &crate::verification::VerificationRequest,
307 ) -> Vec<RequestLogEntry> {
308 let logs = self.logs.read().await;
309 logs.iter()
310 .filter(|entry| crate::verification::matches_verification_pattern(entry, pattern))
311 .cloned()
312 .collect()
313 }
314
315 pub async fn count_matching_requests(
321 &self,
322 pattern: &crate::verification::VerificationRequest,
323 ) -> usize {
324 let logs = self.logs.read().await;
325 logs.iter()
326 .filter(|entry| crate::verification::matches_verification_pattern(entry, pattern))
327 .count()
328 }
329
330 pub async fn get_request_sequence(
336 &self,
337 patterns: &[crate::verification::VerificationRequest],
338 ) -> Vec<RequestLogEntry> {
339 let logs = self.logs.read().await;
340 let mut log_idx = 0;
341 let mut all_matches = Vec::new();
342
343 for pattern in patterns {
344 let mut found = false;
346 while log_idx < logs.len() {
347 if crate::verification::matches_verification_pattern(&logs[log_idx], pattern) {
348 all_matches.push(logs[log_idx].clone());
349 log_idx += 1;
350 found = true;
351 break;
352 }
353 log_idx += 1;
354 }
355
356 if !found {
357 break;
359 }
360 }
361
362 all_matches
363 }
364}
365
366static GLOBAL_LOGGER: once_cell::sync::OnceCell<CentralizedRequestLogger> =
368 once_cell::sync::OnceCell::new();
369
370pub fn init_global_logger(max_logs: usize) -> &'static CentralizedRequestLogger {
372 GLOBAL_LOGGER.get_or_init(|| CentralizedRequestLogger::new(max_logs))
373}
374
375pub fn get_global_logger() -> Option<&'static CentralizedRequestLogger> {
377 GLOBAL_LOGGER.get()
378}
379
380pub async fn log_request_global(entry: RequestLogEntry) {
382 if let Some(logger) = get_global_logger() {
383 logger.log_request(entry).await;
384 }
385}
386
387#[allow(clippy::too_many_arguments)]
389pub fn create_http_log_entry(
390 method: &str,
391 path: &str,
392 status_code: u16,
393 response_time_ms: u64,
394 client_ip: Option<String>,
395 user_agent: Option<String>,
396 headers: HashMap<String, String>,
397 response_size_bytes: u64,
398 error_message: Option<String>,
399) -> RequestLogEntry {
400 create_http_log_entry_with_query(
401 method,
402 path,
403 status_code,
404 response_time_ms,
405 client_ip,
406 user_agent,
407 headers,
408 HashMap::new(), response_size_bytes,
410 error_message,
411 )
412}
413
414#[allow(clippy::too_many_arguments)]
416pub fn create_http_log_entry_with_query(
417 method: &str,
418 path: &str,
419 status_code: u16,
420 response_time_ms: u64,
421 client_ip: Option<String>,
422 user_agent: Option<String>,
423 headers: HashMap<String, String>,
424 query_params: HashMap<String, String>,
425 response_size_bytes: u64,
426 error_message: Option<String>,
427) -> RequestLogEntry {
428 RequestLogEntry {
429 id: uuid::Uuid::new_v4().to_string(),
430 timestamp: Utc::now(),
431 server_type: "HTTP".to_string(),
432 method: method.to_string(),
433 path: path.to_string(),
434 status_code,
435 response_time_ms,
436 client_ip,
437 user_agent,
438 headers,
439 query_params,
440 response_size_bytes,
441 error_message,
442 metadata: HashMap::new(),
443 reality_metadata: None,
444 }
445}
446
447pub fn create_websocket_log_entry(
449 event_type: &str, path: &str,
451 status_code: u16,
452 client_ip: Option<String>,
453 message_size_bytes: u64,
454 error_message: Option<String>,
455) -> RequestLogEntry {
456 let mut metadata = HashMap::new();
457 metadata.insert("event_type".to_string(), event_type.to_string());
458
459 RequestLogEntry {
460 id: uuid::Uuid::new_v4().to_string(),
461 timestamp: Utc::now(),
462 server_type: "WebSocket".to_string(),
463 method: event_type.to_uppercase(),
464 path: path.to_string(),
465 status_code,
466 response_time_ms: 0, client_ip,
468 user_agent: None,
469 headers: HashMap::new(),
470 query_params: HashMap::new(),
471 response_size_bytes: message_size_bytes,
472 error_message,
473 metadata,
474 reality_metadata: None,
475 }
476}
477
478#[allow(clippy::too_many_arguments)]
480pub fn create_grpc_log_entry(
481 service: &str,
482 method: &str,
483 status_code: u16, response_time_ms: u64,
485 client_ip: Option<String>,
486 request_size_bytes: u64,
487 response_size_bytes: u64,
488 error_message: Option<String>,
489) -> RequestLogEntry {
490 let mut metadata = HashMap::new();
491 metadata.insert("service".to_string(), service.to_string());
492 metadata.insert("request_size_bytes".to_string(), request_size_bytes.to_string());
493
494 RequestLogEntry {
495 id: uuid::Uuid::new_v4().to_string(),
496 timestamp: Utc::now(),
497 server_type: "gRPC".to_string(),
498 method: format!("{}/{}", service, method),
499 path: format!("/{}/{}", service, method),
500 status_code,
501 response_time_ms,
502 client_ip,
503 user_agent: None,
504 headers: HashMap::new(),
505 query_params: HashMap::new(),
506 response_size_bytes,
507 error_message,
508 metadata,
509 reality_metadata: None,
510 }
511}
512
513#[cfg(test)]
514mod tests {
515 use super::*;
516
517 fn create_test_entry(server_type: &str, method: &str) -> RequestLogEntry {
518 RequestLogEntry {
519 id: uuid::Uuid::new_v4().to_string(),
520 timestamp: Utc::now(),
521 server_type: server_type.to_string(),
522 method: method.to_string(),
523 path: "/test".to_string(),
524 status_code: 200,
525 response_time_ms: 100,
526 client_ip: Some("127.0.0.1".to_string()),
527 user_agent: Some("test-agent".to_string()),
528 headers: HashMap::new(),
529 response_size_bytes: 1024,
530 error_message: None,
531 metadata: HashMap::new(),
532 reality_metadata: None,
533 }
534 }
535
536 #[test]
537 fn test_centralized_logger_new() {
538 let logger = CentralizedRequestLogger::new(500);
539 assert_eq!(logger.max_logs, 500);
540 }
541
542 #[test]
543 fn test_centralized_logger_default() {
544 let logger = CentralizedRequestLogger::default();
545 assert_eq!(logger.max_logs, 1000);
546 }
547
548 #[tokio::test]
549 async fn test_log_request() {
550 let logger = CentralizedRequestLogger::new(10);
551 let entry = create_test_entry("HTTP", "GET");
552
553 logger.log_request(entry).await;
554
555 let logs = logger.get_recent_logs(None).await;
556 assert_eq!(logs.len(), 1);
557 assert_eq!(logs[0].method, "GET");
558 }
559
560 #[tokio::test]
561 async fn test_log_request_maintains_size_limit() {
562 let logger = CentralizedRequestLogger::new(5);
563
564 for i in 0..10 {
566 let mut entry = create_test_entry("HTTP", "GET");
567 entry.id = format!("entry-{}", i);
568 logger.log_request(entry).await;
569 }
570
571 let logs = logger.get_recent_logs(None).await;
572 assert_eq!(logs.len(), 5); }
574
575 #[tokio::test]
576 async fn test_get_recent_logs_with_limit() {
577 let logger = CentralizedRequestLogger::new(100);
578
579 for _ in 0..20 {
580 logger.log_request(create_test_entry("HTTP", "GET")).await;
581 }
582
583 let logs = logger.get_recent_logs(Some(10)).await;
584 assert_eq!(logs.len(), 10);
585 }
586
587 #[tokio::test]
588 async fn test_get_logs_by_server() {
589 let logger = CentralizedRequestLogger::new(100);
590
591 logger.log_request(create_test_entry("HTTP", "GET")).await;
592 logger.log_request(create_test_entry("HTTP", "POST")).await;
593 logger.log_request(create_test_entry("WebSocket", "CONNECT")).await;
594 logger.log_request(create_test_entry("gRPC", "Call")).await;
595
596 let http_logs = logger.get_logs_by_server("HTTP", None).await;
597 assert_eq!(http_logs.len(), 2);
598
599 let ws_logs = logger.get_logs_by_server("WebSocket", None).await;
600 assert_eq!(ws_logs.len(), 1);
601
602 let grpc_logs = logger.get_logs_by_server("gRPC", None).await;
603 assert_eq!(grpc_logs.len(), 1);
604 }
605
606 #[tokio::test]
607 async fn test_get_request_counts_by_server() {
608 let logger = CentralizedRequestLogger::new(100);
609
610 logger.log_request(create_test_entry("HTTP", "GET")).await;
611 logger.log_request(create_test_entry("HTTP", "POST")).await;
612 logger.log_request(create_test_entry("HTTP", "PUT")).await;
613 logger.log_request(create_test_entry("WebSocket", "CONNECT")).await;
614 logger.log_request(create_test_entry("gRPC", "Call")).await;
615 logger.log_request(create_test_entry("gRPC", "Stream")).await;
616
617 let counts = logger.get_request_counts_by_server().await;
618
619 assert_eq!(counts.get("HTTP"), Some(&3));
620 assert_eq!(counts.get("WebSocket"), Some(&1));
621 assert_eq!(counts.get("gRPC"), Some(&2));
622 }
623
624 #[tokio::test]
625 async fn test_clear_logs() {
626 let logger = CentralizedRequestLogger::new(100);
627
628 logger.log_request(create_test_entry("HTTP", "GET")).await;
629 logger.log_request(create_test_entry("HTTP", "POST")).await;
630
631 let logs = logger.get_recent_logs(None).await;
632 assert_eq!(logs.len(), 2);
633
634 logger.clear_logs().await;
635
636 let logs = logger.get_recent_logs(None).await;
637 assert_eq!(logs.len(), 0);
638 }
639
640 #[test]
641 fn test_create_http_log_entry() {
642 let mut headers = HashMap::new();
643 headers.insert("Content-Type".to_string(), "application/json".to_string());
644
645 let entry = create_http_log_entry(
646 "POST",
647 "/api/test",
648 201,
649 150,
650 Some("192.168.1.1".to_string()),
651 Some("Mozilla/5.0".to_string()),
652 headers.clone(),
653 2048,
654 None,
655 );
656
657 assert_eq!(entry.server_type, "HTTP");
658 assert_eq!(entry.method, "POST");
659 assert_eq!(entry.path, "/api/test");
660 assert_eq!(entry.status_code, 201);
661 assert_eq!(entry.response_time_ms, 150);
662 assert_eq!(entry.response_size_bytes, 2048);
663 assert_eq!(entry.client_ip, Some("192.168.1.1".to_string()));
664 assert_eq!(entry.user_agent, Some("Mozilla/5.0".to_string()));
665 assert_eq!(entry.headers.get("Content-Type"), Some(&"application/json".to_string()));
666 assert!(entry.error_message.is_none());
667 }
668
669 #[test]
670 fn test_create_websocket_log_entry() {
671 let entry = create_websocket_log_entry(
672 "connect",
673 "/ws/chat",
674 101,
675 Some("10.0.0.1".to_string()),
676 0,
677 None,
678 );
679
680 assert_eq!(entry.server_type, "WebSocket");
681 assert_eq!(entry.method, "CONNECT");
682 assert_eq!(entry.path, "/ws/chat");
683 assert_eq!(entry.status_code, 101);
684 assert_eq!(entry.response_time_ms, 0);
685 assert_eq!(entry.metadata.get("event_type"), Some(&"connect".to_string()));
686 }
687
688 #[test]
689 fn test_create_grpc_log_entry() {
690 let entry = create_grpc_log_entry(
691 "UserService",
692 "GetUser",
693 0, 50,
695 Some("172.16.0.1".to_string()),
696 128,
697 512,
698 None,
699 );
700
701 assert_eq!(entry.server_type, "gRPC");
702 assert_eq!(entry.method, "UserService/GetUser");
703 assert_eq!(entry.path, "/UserService/GetUser");
704 assert_eq!(entry.status_code, 0);
705 assert_eq!(entry.response_time_ms, 50);
706 assert_eq!(entry.response_size_bytes, 512);
707 assert_eq!(entry.metadata.get("service"), Some(&"UserService".to_string()));
708 assert_eq!(entry.metadata.get("request_size_bytes"), Some(&"128".to_string()));
709 }
710
711 #[test]
712 fn test_request_log_entry_with_error() {
713 let entry = create_http_log_entry(
714 "GET",
715 "/api/error",
716 500,
717 200,
718 None,
719 None,
720 HashMap::new(),
721 0,
722 Some("Internal server error".to_string()),
723 );
724
725 assert_eq!(entry.status_code, 500);
726 assert_eq!(entry.error_message, Some("Internal server error".to_string()));
727 }
728}