feat: implement structured logging with correlation IDs and centralized aggregation

backend/alerting/logAlerts.ts

@@ -0,0 +1,44 @@
+import { LogEntry, LogStorage } from '../elasticsearch/logStorage';
+
+// Threshold: 10 errors within 5 minutes triggers an alert
+const ERROR_THRESHOLD = 10;
+const TIME_WINDOW_MS = 5 * 60 * 1000;
+
+let lastAlertTime = 0;
+const ALERT_COOLDOWN_MS = 15 * 60 * 1000; // 15 mins cooldown
+
+export async function checkLogAlerts(newLog: LogEntry, storage: LogStorage) {
+  if (newLog.level !== 'error') return;
+
+  const now = Date.now();
+  if (now - lastAlertTime < ALERT_COOLDOWN_MS) {
+    return; // Cooldown active
+  }
+
+  const windowStart = new Date(now - TIME_WINDOW_MS);
+
+  // Search recent errors
+  const recentErrors = await storage.searchLogs({
+    level: 'error',
+    startDate: windowStart,
+    limit: 100,
+  });
+
+  if (recentErrors.total >= ERROR_THRESHOLD) {
+    triggerAlert(`High Error Rate Detected: ${recentErrors.total} errors in the last 5 minutes.`);
+    lastAlertTime = now;
+  }
+
+  // Pattern matching for critical errors
+  const criticalPatterns = ['payment failed', 'database connection lost', 'out of memory'];
+  if (criticalPatterns.some(pattern => newLog.message.toLowerCase().includes(pattern))) {
+    triggerAlert(`CRITICAL ERROR DETECTED: ${newLog.message}`);
+    // Update cooldown to prevent spam if critical error repeats rapidly
+    lastAlertTime = now;
+  }
+}
+
+function triggerAlert(message: string) {
+  // In a real system, this would integrate with PagerDuty, Slack, Email, etc.
+  console.warn(`\n[ALERT] 🚨🚨🚨 ${message} 🚨🚨🚨\n`);
+}

backend/elasticsearch/logStorage.ts

@@ -0,0 +1,118 @@
+import AsyncStorage from '@react-native-async-storage/async-storage';
+import { LogLevel } from '../services/shared/logging';
+import { checkLogAlerts } from '../alerting/logAlerts';
+
+export interface LogEntry {
+  id: string;
+  level: LogLevel;
+  message: string;
+  timestamp: string;
+  correlationId?: string;
+  [key: string]: any;
+}
+
+export interface LogSearchQuery {
+  level?: LogLevel;
+  correlationId?: string;
+  searchTerm?: string;
+  startDate?: Date;
+  endDate?: Date;
+  limit?: number;
+  offset?: number;
+}
+
+const LOG_STORAGE_KEY = '@subtrackr/elasticsearch_logs';
+const MAX_LOGS = 10000;
+
+export class LogStorage {
+  private logs: LogEntry[] = [];
+  private initialized = false;
+
+  async init() {
+    if (this.initialized) return;
+    try {
+      const storedLogs = await AsyncStorage.getItem(LOG_STORAGE_KEY);
+      if (storedLogs) {
+        this.logs = JSON.parse(storedLogs);
+      }
+    } catch (e) {
+      console.error('Failed to load logs from storage', e);
+    }
+    this.initialized = true;
+  }
+
+  private async persist() {
+    try {
+      await AsyncStorage.setItem(LOG_STORAGE_KEY, JSON.stringify(this.logs));
+    } catch (e) {
+      console.error('Failed to persist logs', e);
+    }
+  }
+
+  async insertLog(entry: LogEntry) {
+    if (!this.initialized) await this.init();
+
+    this.logs.unshift(entry); // Add to beginning
+
+    // Check for alerts
+    await checkLogAlerts(entry, this);
+
+    if (this.logs.length > MAX_LOGS) {
+      this.logs = this.logs.slice(0, MAX_LOGS);
+    }
+
+    // Debounce or periodic persist in a real app, here we do it immediately
+    void this.persist();
+  }
+
+  async searchLogs(query: LogSearchQuery): Promise<{ data: LogEntry[]; total: number }> {
+    if (!this.initialized) await this.init();
+
+    let filtered = this.logs;
+
+    if (query.level) {
+      filtered = filtered.filter(l => l.level === query.level);
+    }
+
+    if (query.correlationId) {
+      filtered = filtered.filter(l => l.correlationId === query.correlationId);
+    }
+
+    if (query.startDate) {
+      filtered = filtered.filter(l => new Date(l.timestamp) >= query.startDate!);
+    }
+
+    if (query.endDate) {
+      filtered = filtered.filter(l => new Date(l.timestamp) <= query.endDate!);
+    }
+
+    if (query.searchTerm) {
+      const term = query.searchTerm.toLowerCase();
+      filtered = filtered.filter(l => 
+        l.message.toLowerCase().includes(term) || 
+        JSON.stringify(l).toLowerCase().includes(term)
+      );
+    }
+
+    const total = filtered.length;
+    const offset = query.offset || 0;
+    const limit = query.limit || 50;
+
+    return {
+      data: filtered.slice(offset, offset + limit),
+      total
+    };
+  }
+
+  async cleanupOldLogs(daysToKeep: number = 30) {
+    if (!this.initialized) await this.init();
+
+    const cutoffDate = new Date();
+    cutoffDate.setDate(cutoffDate.getDate() - daysToKeep);
+
+    this.logs = this.logs.filter(l => new Date(l.timestamp) >= cutoffDate);
+    await this.persist();
+  }
+}
+
+export const logStorage = new LogStorage();

backend/ml/churnModel.py

@@ -1,6 +1,7 @@
 import math
 import random
 from typing import Dict, List, Optional
+from .logger import logger
 
 class ChurnPredictionModel:
     def __init__(self):
@@ -36,10 +37,11 @@ def _extract_features(self, user_data: Dict) -> Dict:
 
         return features
 
-    def predict_churn(self, subscriber_address: str, user_data: Dict) -> Dict:
+    def predict_churn(self, subscriber_address: str, user_data: Dict, correlation_id: str = None) -> Dict:
         """
         Predict churn probability and return risk scoring.
         """
+        logger.info(f"Predicting churn for {subscriber_address}", correlation_id=correlation_id)
         features = self._extract_features(user_data)
 
         # Calculate risk score (0.0 to 1.0)
@@ -62,13 +64,15 @@ def predict_churn(self, subscriber_address: str, user_data: Dict) -> Dict:
             for factor in sorted_factors if factor[1] > 0.1
         ]
 
-        return {
+        result = {
             "subscriber": subscriber_address,
             "churn_probability": round(risk_score, 4),
             "risk_level": risk_level,
             "risk_factors": top_factors,
             "recommended_action": self._get_recommended_action(risk_level, top_factors)
         }
+        logger.debug(f"Churn prediction result for {subscriber_address}: {risk_score}", correlation_id=correlation_id)
+        return result
 
     def _get_recommended_action(self, risk_level: str, top_factors: List[Dict]) -> str:
         if risk_level == "Low":

backend/ml/logger.py

@@ -0,0 +1,36 @@
+import json
+import uuid
+import datetime
+
+class StructuredLogger:
+    def __init__(self, name="ml-service"):
+        self.name = name
+
+    def _log(self, level, message, correlation_id=None, **kwargs):
+        log_entry = {
+            "timestamp": datetime.datetime.utcnow().isoformat() + "Z",
+            "level": level,
+            "message": message,
+            "service": self.name
+        }
+        if correlation_id:
+            log_entry["correlationId"] = correlation_id
+
+        if kwargs:
+            log_entry.update(kwargs)
+
+        print(json.dumps(log_entry))
+
+    def debug(self, message, correlation_id=None, **kwargs):
+        self._log("debug", message, correlation_id, **kwargs)
+
+    def info(self, message, correlation_id=None, **kwargs):
+        self._log("info", message, correlation_id, **kwargs)
+
+    def warn(self, message, correlation_id=None, **kwargs):
+        self._log("warn", message, correlation_id, **kwargs)
+
+    def error(self, message, correlation_id=None, **kwargs):
+        self._log("error", message, correlation_id, **kwargs)
+
+logger = StructuredLogger()

backend/ml/pricingModel.py

@@ -1,6 +1,7 @@
 import math
 import random
 from typing import Dict, List, Optional
+from .logger import logger
 
 class PricingOptimizationEngine:
     def __init__(self):
@@ -28,7 +29,7 @@ def estimate_willingness_to_pay(self, usage_data: Dict) -> float:
         wtp = base_wtp * (1 + (usage_frequency * 0.05) + (retention_rate * 0.2))
         return round(wtp, 2)
 
-    def calculate_optimal_price(self, subscription_id: str, context: Dict) -> Dict:
+    def calculate_optimal_price(self, subscription_id: str, context: Dict, correlation_id: str = None) -> Dict:
         """
         Calculates the optimal price based on several factors.
         """
@@ -57,8 +58,10 @@ def calculate_optimal_price(self, subscription_id: str, context: Dict) -> Dict:
             },
             "recommendation": "Increase" if optimal_price > current_price else "Decrease" if optimal_price < current_price else "Maintain"
         }
+        logger.info(f"Calculated optimal price {optimal_price} for {subscription_id}", correlation_id=correlation_id)
+        return result
 
-    def get_price_recommendations(self, plan_id: str, historical_data: List[Dict]) -> List[Dict]:
+    def get_price_recommendations(self, plan_id: str, historical_data: List[Dict], correlation_id: str = None) -> List[Dict]:
         """
         Returns a range of price recommendations for a specific plan.
         Useful for A/B testing setup.

backend/ml/recommendationModel.py

@@ -1,6 +1,7 @@
 import math
 import random
 from typing import Dict, List, Optional
+from .logger import logger
 
 class RecommendationEngine:
     def __init__(self):
@@ -39,7 +40,7 @@ def _content_based_filtering(self, subscriber_address: str, user_profile: Dict)
                 recommendations.append({"id": rec_id, "score": 0.88})
         return recommendations
 
-    def get_recommendations(self, subscriber_address: str, context: Dict) -> List[Dict]:
+    def get_recommendations(self, subscriber_address: str, context: Dict, correlation_id: str = None) -> List[Dict]:
         """
         Combines collaborative and content-based filtering.
         """
@@ -81,6 +82,7 @@ def get_recommendations(self, subscriber_address: str, context: Dict) -> List[Di
                 "confidence_score": 0.50
             })
 
+        logger.info(f"Generated {len(final_recs)} recommendations for {subscriber_address}", correlation_id=correlation_id)
         return final_recs
 
 if __name__ == "__main__":

backend/services/shared/logging.ts

@@ -1,7 +1,15 @@
 import { piiClassifier, type ClassificationLevel } from './piiClassifier';
+import { AsyncLocalStorage } from 'node:async_hooks';
+import { logStorage } from '../../elasticsearch/logStorage';
 
 export type LogLevel = 'debug' | 'info' | 'warn' | 'error';
 
+export const correlationIdStorage = new AsyncLocalStorage<string>();
+
+export function withCorrelationId<T>(correlationId: string, fn: () => T): T {
+  return correlationIdStorage.run(correlationId, fn);
+}
+
 const LOG_LEVEL_PRIORITY: Record<LogLevel, number> = {
   debug: 0,
   info: 1,
@@ -54,25 +62,30 @@ function formatLog(level: LogLevel, message: string, context?: LogContext) {
 }
 
 function sendToConsole(logEntry: any) {
-  console.log(JSON.stringify(logEntry, null, 2));
+  console.log(JSON.stringify(logEntry));
 }
 
-//  future: plug Sentry / API here
-async function sendToRemote(_logEntry: any) {
-  // Example:
-  // await fetch("https://your-api/logs", { method: "POST", body: JSON.stringify(logEntry) });
+async function sendToRemote(logEntry: any) {
+  try {
+    await logStorage.insertLog(logEntry);
+  } catch (e) {
+    console.error('Failed to forward log to logStorage', e);
+  }
 }
 
 function log(level: LogLevel, message: string, context?: LogContext) {
   if (!shouldLog(level)) return;
 
-  const logEntry = formatLog(level, message, sanitizeContext(context));
+  const currentCorrelationId = correlationIdStorage.getStore();
+  const mergedContext = {
+    correlationId: currentCorrelationId,
+    ...context
+  };
 
-  sendToConsole(logEntry);
+  const logEntry = formatLog(level, message, sanitizeContext(mergedContext));
 
-  if (level === 'error') {
-    sendToRemote(logEntry);
-  }
+  sendToConsole(logEntry);
+  void sendToRemote(logEntry);
 }
 
 export const logger = {

contracts/src/logging.rs

@@ -0,0 +1,37 @@
+use soroban_sdk::{Env, Symbol, String, val};
+
+/// Standard structured log event for SubTrackr contracts.
+/// By emitting these events, the off-chain indexing services can aggregate
+/// contract logs and correlate them with backend requests using correlation_id.
+pub struct ContractLogger;
+
+impl ContractLogger {
+    /// Log an informational event with an optional correlation_id
+    pub fn info(env: &Env, message: &str, correlation_id: Option<String>) {
+        Self::log(env, "info", message, correlation_id);
+    }
+
+    /// Log a warning event with an optional correlation_id
+    pub fn warn(env: &Env, message: &str, correlation_id: Option<String>) {
+        Self::log(env, "warn", message, correlation_id);
+    }
+
+    /// Log an error event with an optional correlation_id
+    pub fn error(env: &Env, message: &str, correlation_id: Option<String>) {
+        Self::log(env, "error", message, correlation_id);
+    }
+
+    fn log(env: &Env, level: &str, message: &str, correlation_id: Option<String>) {
+        let topics = (Symbol::new(env, "structured_log"), Symbol::new(env, level));
+
+        // Structure the log data
+        let mut data_map = soroban_sdk::Map::new(env);
+        data_map.set(Symbol::new(env, "msg"), String::from_str(env, message));
+
+        if let Some(cid) = correlation_id {
+            data_map.set(Symbol::new(env, "correlation_id"), cid);
+        }
+
+        env.events().publish(topics, data_map);
+    }
+}