Real-Time IP Intelligence Overview

// Production-ready real-time IP intelligence processing engine
interface IPIntelligenceEvent {
  ip: string
  timestamp: number
  eventType: 'location_update' | 'threat_detected' | 'behavior_anomaly' | 'reputation_change'
  confidence: number
  data: {
    geolocation?: GeolocationResult
    threatScore?: number
    reputationScore?: number
    behavioralFlags?: string[]
    metadata?: Record<string, any>
  }
  source: string
  priority: 'low' | 'medium' | 'high' | 'critical'
}

interface IPIntelligenceProfile {
  ip: string
  currentLocation?: GeolocationResult
  locationHistory: GeolocationResult[]
  threatScore: number
  reputationScore: number
  behavioralProfile: {
    patterns: string[]
    riskFactors: string[]
    lastActivity: number
    activityFrequency: number
  }
  alerts: Array<{
    type: string
    severity: string
    timestamp: number
    description: string
  }>
  lastUpdated: number
  confidence: number
}

interface StreamingConfig {
  batchSize: number
  flushInterval: number
  maxQueueSize: number
  retryAttempts: number
  backoffMultiplier: number
}

interface AnalyticsMetrics {
  processedEvents: number
  averageProcessingTime: number
  errorRate: number
  throughput: number // events per second
  queueSize: number
  memoryUsage: number
}

class RealTimeIPIntelligenceEngine {
  private eventQueue: IPIntelligenceEvent[] = []
  private profiles: Map<string, IPIntelligenceProfile> = new Map()
  private subscribers: Map<string, Array<(event: IPIntelligenceEvent) => void>> = new Map()
  private config: StreamingConfig
  private metrics: AnalyticsMetrics
  private isProcessing = false

  constructor(config: StreamingConfig) {
    this.config = config
    this.metrics = {
      processedEvents: 0,
      averageProcessingTime: 0,
      errorRate: 0,
      throughput: 0,
      queueSize: 0,
      memoryUsage: 0
    }

    this.startProcessingLoop()
    this.startMetricsCollection()
  }

  // Process incoming IP intelligence events
  async processEvent(event: IPIntelligenceEvent): Promise<void> {
    return new Promise((resolve) => {
      event.timestamp = Date.now()
      this.eventQueue.push(event)

      if (this.eventQueue.length >= this.config.batchSize) {
        this.flushQueue()
      }

      resolve()
    })
  }

  // Subscribe to real-time intelligence updates
  subscribe(eventType: string, callback: (event: IPIntelligenceEvent) => void): string {
    const subscriptionId = `sub_${Date.now()}_${Math.random()}`

    if (!this.subscribers.has(eventType)) {
      this.subscribers.set(eventType, [])
    }

    this.subscribers.get(eventType)!.push(callback)

    return subscriptionId
  }

  // Unsubscribe from intelligence updates
  unsubscribe(eventType: string, subscriptionId: string): void {
    const callbacks = this.subscribers.get(eventType)
    if (callbacks) {
      const index = callbacks.findIndex(cb => cb.name === subscriptionId)
      if (index > -1) {
        callbacks.splice(index, 1)
      }
    }
  }

  // Get current intelligence profile for IP
  getIntelligenceProfile(ip: string): IPIntelligenceProfile | null {
    return this.profiles.get(ip) || null
  }

  // Get real-time analytics metrics
  getMetrics(): AnalyticsMetrics {
    return { ...this.metrics }
  }

  private async flushQueue(): Promise<void> {
    if (this.eventQueue.length === 0 || this.isProcessing) return

    this.isProcessing = true
    const batch = this.eventQueue.splice(0, this.config.batchSize)

    try {
      await this.processBatch(batch)
      this.updateMetrics(batch.length, Date.now())
    } catch (error) {
      console.error('Error processing intelligence batch:', error)
      this.handleProcessingError(batch, error)
    } finally {
      this.isProcessing = false
    }
  }

  private async processBatch(events: IPIntelligenceEvent[]): Promise<void> {
    const startTime = Date.now()

    for (const event of events) {
      await this.processSingleEvent(event)
    }

    const processingTime = Date.now() - startTime
    this.updateAverageProcessingTime(processingTime)
  }

  private async processSingleEvent(event: IPIntelligenceEvent): Promise<void> {
    // Update or create intelligence profile
    let profile = this.profiles.get(event.ip)

    if (!profile) {
      profile = this.createNewProfile(event.ip)
      this.profiles.set(event.ip, profile)
    }

    // Process event based on type
    switch (event.eventType) {
      case 'location_update':
        await this.processLocationUpdate(event, profile)
        break
      case 'threat_detected':
        await this.processThreatDetection(event, profile)
        break
      case 'behavior_anomaly':
        await this.processBehavioralAnomaly(event, profile)
        break
      case 'reputation_change':
        await this.processReputationChange(event, profile)
        break
    }

    // Notify subscribers
    this.notifySubscribers(event)

    // Update profile timestamp
    profile.lastUpdated = Date.now()
  }

  private createNewProfile(ip: string): IPIntelligenceProfile {
    return {
      ip,
      locationHistory: [],
      threatScore: 50, // Neutral starting score
      reputationScore: 50,
      behavioralProfile: {
        patterns: [],
        riskFactors: [],
        lastActivity: Date.now(),
        activityFrequency: 0
      },
      alerts: [],
      lastUpdated: Date.now(),
      confidence: 0
    }
  }

  private async processLocationUpdate(event: IPIntelligenceEvent, profile: IPIntelligenceProfile): Promise<void> {
    if (event.data.geolocation) {
      profile.currentLocation = event.data.geolocation
      profile.locationHistory.push(event.data.geolocation)

      // Keep only last 100 location records
      if (profile.locationHistory.length > 100) {
        profile.locationHistory = profile.locationHistory.slice(-100)
      }

      // Calculate location-based threat score
      profile.threatScore = this.calculateLocationThreatScore(event.data.geolocation, profile)
    }
  }

  private async processThreatDetection(event: IPIntelligenceEvent, profile: IPIntelligenceProfile): Promise<void> {
    if (event.data.threatScore !== undefined) {
      profile.threatScore = Math.max(profile.threatScore, event.data.threatScore)

      // Add alert if threat score is high
      if (event.data.threatScore > 75) {
        profile.alerts.push({
          type: 'threat_detected',
          severity: event.data.threatScore > 90 ? 'critical' : 'high',
          timestamp: Date.now(),
          description: `Threat score ${event.data.threatScore} detected for IP ${event.ip}`
        })
      }
    }
  }

  private async processBehavioralAnomaly(event: IPIntelligenceEvent, profile: IPIntelligenceProfile): Promise<void> {
    if (event.data.behavioralFlags) {
      profile.behavioralProfile.riskFactors.push(...event.data.behavioralFlags)
      profile.behavioralProfile.lastActivity = Date.now()

      // Update activity frequency
      const recentActivity = profile.locationHistory.filter(
        loc => Date.now() - loc.timestamp < 24 * 60 * 60 * 1000 // Last 24 hours
      )
      profile.behavioralProfile.activityFrequency = recentActivity.length

      // Generate alert for significant anomalies
      if (event.data.behavioralFlags.some(flag => flag.includes('suspicious'))) {
        profile.alerts.push({
          type: 'behavioral_anomaly',
          severity: 'medium',
          timestamp: Date.now(),
          description: `Behavioral anomaly detected: ${event.data.behavioralFlags.join(', ')}`
        })
      }
    }
  }

  private async processReputationChange(event: IPIntelligenceEvent, profile: IPIntelligenceProfile): Promise<void> {
    if (event.data.reputationScore !== undefined) {
      profile.reputationScore = event.data.reputationScore

      // Add alert if reputation drops significantly
      const reputationDrop = profile.reputationScore - (profile.reputationScore + event.data.reputationScore) / 2
      if (reputationDrop < -20) {
        profile.alerts.push({
          type: 'reputation_change',
          severity: reputationDrop < -50 ? 'critical' : 'high',
          timestamp: Date.now(),
          description: `Reputation score dropped by ${Math.abs(reputationDrop)} points`
        })
      }
    }
  }

  private calculateLocationThreatScore(location: GeolocationResult, profile: IPIntelligenceProfile): number {
    let score = 50 // Base score

    // VPN/Proxy detection reduces score
    if (location.metadata?.isVpn || location.metadata?.isDatacenter) {
      score -= 20
    }

    // Corporate networks increase score
    if (location.metadata?.isCorporate) {
      score += 10
    }

    // High-risk countries
    const highRiskCountries = ['RU', 'CN', 'IR', 'KP', 'SY']
    if (highRiskCountries.includes(location.countryCode)) {
      score -= 15
    }

    // Recent location changes
    if (profile.locationHistory.length > 1) {
      const lastLocation = profile.locationHistory[profile.locationHistory.length - 2]
      const timeDiff = (Date.now() - lastLocation.timestamp) / (1000 * 3600) // hours
      const distance = this.calculateDistance(
        lastLocation.latitude || 0,
        lastLocation.longitude || 0,
        location.latitude || 0,
        location.longitude || 0
      )

      if (distance > 1000 && timeDiff < 2) { // 1000km in less than 2 hours
        score -= 25 // Suspicious rapid movement
      }
    }

    return Math.max(0, Math.min(100, score))
  }

  private calculateDistance(lat1: number, lng1: number, lat2: number, lng2: number): number {
    const R = 6371 // Earth's radius in km
    const dLat = this.toRadians(lat2 - lat1)
    const dLng = this.toRadians(lng2 - lng1)

    const a = Math.sin(dLat / 2) * Math.sin(dLat / 2) +
              Math.cos(this.toRadians(lat1)) * Math.cos(this.toRadians(lat2)) *
              Math.sin(dLng / 2) * Math.sin(dLng / 2)

    const c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a))
    return R * c
  }

  private toRadians(degrees: number): number {
    return degrees * (Math.PI / 180)
  }

  private notifySubscribers(event: IPIntelligenceEvent): void {
    const callbacks = this.subscribers.get(event.eventType) || []
    const profileCallbacks = this.subscribers.get('profile_update') || []

    callbacks.forEach(callback => {
      try {
        callback(event)
      } catch (error) {
        console.error('Error in event subscriber:', error)
      }
    })

    // Notify profile update subscribers
    profileCallbacks.forEach(callback => {
      try {
        callback(event)
      } catch (error) {
        console.error('Error in profile update subscriber:', error)
      }
    })
  }

  private updateMetrics(batchSize: number, processingTime: number): void {
    this.metrics.processedEvents += batchSize
    this.metrics.throughput = (batchSize / (processingTime / 1000)) // events per second
    this.metrics.queueSize = this.eventQueue.length

    // Update memory usage estimate
    this.metrics.memoryUsage = this.profiles.size * 0.5 + this.eventQueue.length * 0.1 // MB estimate
  }

  private updateAverageProcessingTime(processingTime: number): void {
    const alpha = 0.1 // Smoothing factor
    this.metrics.averageProcessingTime =
      this.metrics.averageProcessingTime * (1 - alpha) + processingTime * alpha
  }

  private handleProcessingError(batch: IPIntelligenceEvent[], error: any): void {
    this.metrics.errorRate = (this.metrics.errorRate + batch.length) / (this.metrics.processedEvents + batch.length)

    // Retry failed events with exponential backoff
    setTimeout(() => {
      batch.forEach(event => this.eventQueue.unshift(event))
    }, 1000 * Math.pow(2, this.config.retryAttempts))
  }

  private startProcessingLoop(): void {
    setInterval(async () => {
      if (this.eventQueue.length > 0 && !this.isProcessing) {
        await this.flushQueue()
      }
    }, this.config.flushInterval)
  }

  private startMetricsCollection(): void {
    // Collect system metrics every 30 seconds
    setInterval(() => {
      this.collectSystemMetrics()
    }, 30000)
  }

  private collectSystemMetrics(): void {
    // In production, this would collect actual system metrics
    // For now, we'll simulate some metrics
    const memoryUsage = process.memoryUsage()
    this.metrics.memoryUsage = Math.round(memoryUsage.heapUsed / 1024 / 1024) // MB
  }
}

// Integration with Express.js for real-time API
const intelligenceEngine = new RealTimeIPIntelligenceEngine({
  batchSize: 100,
  flushInterval: 1000, // 1 second
  maxQueueSize: 10000,
  retryAttempts: 3,
  backoffMultiplier: 2
})

// Middleware for automatic IP intelligence processing
const ipIntelligenceMiddleware = async (req: any, res: any, next: any) => {
  try {
    const clientIP = req.ip || req.connection.remoteAddress || req.socket.remoteAddress

    if (clientIP && !clientIP.includes('127.0.0.1') && !clientIP.includes('::1')) {
      // Get current intelligence profile
      const profile = intelligenceEngine.getIntelligenceProfile(clientIP)

      // Create location update event
      const locationEvent: IPIntelligenceEvent = {
        ip: clientIP,
        timestamp: Date.now(),
        eventType: 'location_update',
        confidence: 85,
        data: {
          geolocation: {
            ip: clientIP,
            country: 'United States',
            countryCode: 'US',
            region: 'California',
            city: 'San Francisco',
            latitude: 37.7749,
            longitude: -122.4194,
            accuracy: 'coordinates',
            confidence: 85,
            provider: 'premium_provider',
            cached: false,
            timestamp: Date.now()
          }
        },
        source: 'middleware',
        priority: 'medium'
      }

      // Process event asynchronously
      intelligenceEngine.processEvent(locationEvent).catch(console.error)

      // Add intelligence data to request
      req.ipIntelligence = {
        profile,
        currentLocation: locationEvent.data.geolocation,
        threatScore: profile?.threatScore || 50,
        reputationScore: profile?.reputationScore || 50
      }

      // Set security headers based on intelligence
      if (profile && profile.threatScore > 75) {
        res.set('X-Risk-Level', 'high')
        res.set('X-Threat-Score', profile.threatScore.toString())
      }
    }
  } catch (error) {
    console.error('IP Intelligence middleware error:', error)
    req.ipIntelligence = null
  }

  next()
}

// Real-time intelligence API endpoints
app.use('/api', ipIntelligenceMiddleware)

// Get current IP intelligence
app.get('/api/ip-intelligence/:ip', (req, res) => {
  const profile = intelligenceEngine.getIntelligenceProfile(req.params.ip)

  if (profile) {
    res.json({
      profile,
      metrics: intelligenceEngine.getMetrics(),
      timestamp: new Date().toISOString()
    })
  } else {
    res.status(404).json({
      error: 'IP intelligence profile not found',
      message: 'No intelligence data available for this IP address'
    })
  }
})

// Subscribe to real-time intelligence updates (WebSocket)
app.ws('/api/intelligence/stream', (ws: any) => {
  const subscriptionId = intelligenceEngine.subscribe('profile_update', (event) => {
    ws.send(JSON.stringify({
      type: 'intelligence_update',
      data: event,
      timestamp: new Date().toISOString()
    }))
  })

  ws.on('close', () => {
    intelligenceEngine.unsubscribe('profile_update', subscriptionId)
  })

  ws.on('message', (message: string) => {
    try {
      const data = JSON.parse(message)
      if (data.type === 'subscribe' && data.eventTypes) {
        data.eventTypes.forEach((eventType: string) => {
          intelligenceEngine.subscribe(eventType, (event) => {
            ws.send(JSON.stringify({
              type: 'event',
              eventType,
              data: event,
              timestamp: new Date().toISOString()
            }))
          })
        })
      }
    } catch (error) {
      console.error('WebSocket message error:', error)
    }
  })
})

// Real-time analytics dashboard data
app.get('/api/intelligence/analytics', (req, res) => {
  res.json({
    metrics: intelligenceEngine.getMetrics(),
    activeProfiles: intelligenceEngine['profiles'].size,
    queueSize: intelligenceEngine['eventQueue'].length,
    subscriptions: Array.from(intelligenceEngine['subscribers'].entries()).map(([eventType, callbacks]) => ({
      eventType,
      subscriberCount: callbacks.length
    })),
    timestamp: new Date().toISOString()
  })
})

// Health check with intelligence metrics
app.get('/health/intelligence', (req, res) => {
  const metrics = intelligenceEngine.getMetrics()

  res.json({
    status: metrics.errorRate < 0.01 ? 'healthy' : 'degraded',
    metrics,
    uptime: process.uptime(),
    timestamp: new Date().toISOString()
  })
})

console.log('Real-time IP intelligence engine initialized')

// Advanced streaming analytics for IP intelligence patterns
interface AnalyticsEvent {
  id: string
  timestamp: number
  eventType: 'pattern_detected' | 'trend_analysis' | 'anomaly_alert' | 'correlation_found'
  data: {
    pattern?: {
      type: string
      confidence: number
      duration: number
      affectedIPs: string[]
    }
    trend?: {
      metric: string
      direction: 'increasing' | 'decreasing' | 'stable'
      changeRate: number
      timeframe: number
    }
    anomaly?: {
      type: string
      severity: string
      affectedRegions: string[]
      duration: number
    }
    correlation?: {
      factors: string[]
      strength: number
      sampleSize: number
    }
  }
  metadata: Record<string, any>
}

interface StreamingAnalyticsConfig {
  windowSize: number // Time window in seconds
  slideInterval: number // How often to update analysis
  minPatternSupport: number // Minimum occurrences for pattern detection
  trendSensitivity: number // How sensitive to trend changes (0-1)
  anomalyThreshold: number // Standard deviations for anomaly detection
}

class IPIntelligenceAnalyticsPipeline {
  private eventBuffer: AnalyticsEvent[] = []
  private patternDetector: PatternDetector
  private trendAnalyzer: TrendAnalyzer
  private anomalyDetector: AnomalyDetector
  private correlationEngine: CorrelationEngine
  private config: StreamingAnalyticsConfig

  constructor(config: StreamingAnalyticsConfig) {
    this.config = config
    this.patternDetector = new PatternDetector(config)
    this.trendAnalyzer = new TrendAnalyzer(config)
    this.anomalyDetector = new AnomalyDetector(config)
    this.correlationEngine = new CorrelationEngine(config)

    this.startStreamingAnalysis()
  }

  // Process new analytics event
  async processAnalyticsEvent(event: AnalyticsEvent): Promise<AnalyticsEvent[]> {
    this.eventBuffer.push(event)

    // Keep buffer within window size
    const windowStart = Date.now() - (this.config.windowSize * 1000)
    this.eventBuffer = this.eventBuffer.filter(e => e.timestamp > windowStart)

    const insights: AnalyticsEvent[] = []

    // Run all analytics engines
    const patternInsights = await this.patternDetector.detectPatterns(this.eventBuffer)
    const trendInsights = await this.trendAnalyzer.analyzeTrends(this.eventBuffer)
    const anomalyInsights = await this.anomalyDetector.detectAnomalies(this.eventBuffer)
    const correlationInsights = await this.correlationEngine.findCorrelations(this.eventBuffer)

    insights.push(...patternInsights, ...trendInsights, ...anomalyInsights, ...correlationInsights)

    return insights
  }

  // Get current analytics state
  getAnalyticsState(): {
    bufferSize: number
    windowUtilization: number
    activePatterns: number
    recentTrends: number
    anomalyCount: number
  } {
    return {
      bufferSize: this.eventBuffer.length,
      windowUtilization: this.eventBuffer.length / 1000, // Assuming 1000 is max buffer
      activePatterns: this.patternDetector.getActivePatternCount(),
      recentTrends: this.trendAnalyzer.getActiveTrendCount(),
      anomalyCount: this.anomalyDetector.getAnomalyCount()
    }
  }

  private startStreamingAnalysis(): void {
    setInterval(async () => {
      if (this.eventBuffer.length > 0) {
        const insights = await this.processAnalyticsEvent({
          id: `analysis_${Date.now()}`,
          timestamp: Date.now(),
          eventType: 'trend_analysis',
          data: {},
          metadata: {}
        })

        // Emit insights to subscribers
        this.emitInsights(insights)
      }
    }, this.config.slideInterval * 1000)
  }

  private emitInsights(insights: AnalyticsEvent[]): void {
    // In production, this would emit to WebSocket connections, Kafka, etc.
    insights.forEach(insight => {
      console.log(`Analytics Insight [${insight.eventType}]: ${JSON.stringify(insight.data)}`)
    })
  }
}

class PatternDetector {
  private activePatterns: Map<string, any> = new Map()

  constructor(private config: StreamingAnalyticsConfig) {}

  async detectPatterns(events: AnalyticsEvent[]): Promise<AnalyticsEvent[]> {
    const insights: AnalyticsEvent[] = []
    const patterns = this.findFrequentPatterns(events)

    for (const pattern of patterns) {
      if (pattern.support >= this.config.minPatternSupport) {
        const patternInsight: AnalyticsEvent = {
          id: `pattern_${Date.now()}_${Math.random()}`,
          timestamp: Date.now(),
          eventType: 'pattern_detected',
          data: {
            pattern: {
              type: pattern.type,
              confidence: pattern.confidence,
              duration: pattern.duration,
              affectedIPs: pattern.ips
            }
          },
          metadata: { patternId: pattern.id }
        }

        insights.push(patternInsight)
        this.activePatterns.set(pattern.id, pattern)
      }
    }

    return insights
  }

  private findFrequentPatterns(events: AnalyticsEvent[]): Array<{
    id: string
    type: string
    support: number
    confidence: number
    duration: number
    ips: string[]
  }> {
    // Simplified pattern detection - in production, use more sophisticated algorithms
    const patterns = []

    // Detect location change patterns
    const locationChanges = events.filter(e => e.data.pattern?.type === 'location_change')
    if (locationChanges.length >= 3) {
      patterns.push({
        id: 'rapid_location_changes',
        type: 'location_change',
        support: locationChanges.length,
        confidence: 0.8,
        duration: 300, // 5 minutes
        ips: [...new Set(locationChanges.map(e => e.metadata.ip))]
      })
    }

    // Detect threat spike patterns
    const threatEvents = events.filter(e => e.data.trend?.metric === 'threat_score')
    if (threatEvents.length >= 5) {
      patterns.push({
        id: 'threat_spike',
        type: 'threat_increase',
        support: threatEvents.length,
        confidence: 0.9,
        duration: 180, // 3 minutes
        ips: [...new Set(threatEvents.map(e => e.metadata.ip))]
      })
    }

    return patterns
  }

  getActivePatternCount(): number {
    return this.activePatterns.size
  }
}

class TrendAnalyzer {
  private trendHistory: Map<string, number[]> = new Map()

  constructor(private config: StreamingAnalyticsConfig) {}

  async analyzeTrends(events: AnalyticsEvent[]): Promise<AnalyticsEvent[]> {
    const insights: AnalyticsEvent[] = []

    // Analyze threat score trends
    const threatTrends = this.analyzeMetricTrend(events, 'threat_score')
    threatTrends.forEach(trend => {
      if (Math.abs(trend.changeRate) > this.config.trendSensitivity) {
        insights.push({
          id: `trend_${Date.now()}_${Math.random()}`,
          timestamp: Date.now(),
          eventType: 'trend_analysis',
          data: { trend },
          metadata: { metric: 'threat_score' }
        })
      }
    })

    // Analyze activity frequency trends
    const activityTrends = this.analyzeMetricTrend(events, 'activity_frequency')
    activityTrends.forEach(trend => {
      if (Math.abs(trend.changeRate) > this.config.trendSensitivity) {
        insights.push({
          id: `trend_${Date.now()}_${Math.random()}`,
          timestamp: Date.now(),
          eventType: 'trend_analysis',
          data: { trend },
          metadata: { metric: 'activity_frequency' }
        })
      }
    })

    return insights
  }

  private analyzeMetricTrend(events: AnalyticsEvent[], metric: string): Array<{
    metric: string
    direction: 'increasing' | 'decreasing' | 'stable'
    changeRate: number
    timeframe: number
  }> {
    const metricEvents = events.filter(e => e.data.trend?.metric === metric)

    if (metricEvents.length < 3) return []

    // Simple linear regression for trend detection
    const values = metricEvents.map(e => e.data.trend!.changeRate)
    const trend = this.calculateTrend(values)

    return [{
      metric,
      direction: trend > 0.1 ? 'increasing' : trend < -0.1 ? 'decreasing' : 'stable',
      changeRate: trend,
      timeframe: this.config.windowSize
    }]
  }

  private calculateTrend(values: number[]): number {
    if (values.length < 2) return 0

    let sumX = 0, sumY = 0, sumXY = 0, sumXX = 0
    const n = values.length

    for (let i = 0; i < n; i++) {
      sumX += i
      sumY += values[i]
      sumXY += i * values[i]
      sumXX += i * i
    }

    const slope = (n * sumXY - sumX * sumY) / (n * sumXX - sumX * sumX)
    return slope
  }

  getActiveTrendCount(): number {
    return this.trendHistory.size
  }
}

class AnomalyDetector {
  private baselineStats: Map<string, { mean: number; std: number }> = new Map()
  private anomalyCount = 0

  constructor(private config: StreamingAnalyticsConfig) {}

  async detectAnomalies(events: AnalyticsEvent[]): Promise<AnalyticsEvent[]> {
    const insights: AnalyticsEvent[] = []

    // Detect threat score anomalies
    const threatAnomalies = this.detectMetricAnomalies(events, 'threat_score')
    threatAnomalies.forEach(anomaly => {
      insights.push({
        id: `anomaly_${Date.now()}_${Math.random()}`,
        timestamp: Date.now(),
        eventType: 'anomaly_alert',
        data: { anomaly },
        metadata: { metric: 'threat_score' }
      })
      this.anomalyCount++
    })

    // Detect activity anomalies
    const activityAnomalies = this.detectMetricAnomalies(events, 'activity_frequency')
    activityAnomalies.forEach(anomaly => {
      insights.push({
        id: `anomaly_${Date.now()}_${Math.random()}`,
        timestamp: Date.now(),
        eventType: 'anomaly_alert',
        data: { anomaly },
        metadata: { metric: 'activity_frequency' }
      })
      this.anomalyCount++
    })

    return insights
  }

  private detectMetricAnomalies(events: AnalyticsEvent[], metric: string): Array<{
    type: string
    severity: string
    affectedRegions: string[]
    duration: number
  }> {
    const metricEvents = events.filter(e => e.metadata.metric === metric)
    const values = metricEvents.map(e => e.data.anomaly?.severity === 'high' ? 1 : 0)

    if (values.length < 10) return []

    const mean = values.reduce((a, b) => a + b, 0) / values.length
    const variance = values.reduce((acc, val) => acc + Math.pow(val - mean, 2), 0) / values.length
    const std = Math.sqrt(variance)

    // Update baseline statistics
    this.baselineStats.set(metric, { mean, std })

    const anomalies = []
    const currentValue = values[values.length - 1]

    if (Math.abs(currentValue - mean) > this.config.anomalyThreshold * std) {
      anomalies.push({
        type: `${metric}_spike`,
        severity: currentValue > mean + 2 * std ? 'critical' : 'high',
        affectedRegions: ['global'], // In production, derive from events
        duration: 60 // 1 minute
      })
    }

    return anomalies
  }

  getAnomalyCount(): number {
    return this.anomalyCount
  }
}

class CorrelationEngine {
  constructor(private config: StreamingAnalyticsConfig) {}

  async findCorrelations(events: AnalyticsEvent[]): Promise<AnalyticsEvent[]> {
    const insights: AnalyticsEvent[] = []

    // Find correlations between threat scores and geographic regions
    const geoThreatCorrelations = this.findGeographicCorrelations(events)
    geoThreatCorrelations.forEach(correlation => {
      insights.push({
        id: `correlation_${Date.now()}_${Math.random()}`,
        timestamp: Date.now(),
        eventType: 'correlation_found',
        data: { correlation },
        metadata: { correlationType: 'geo_threat' }
      })
    })

    // Find correlations between time of day and activity patterns
    const temporalCorrelations = this.findTemporalCorrelations(events)
    temporalCorrelations.forEach(correlation => {
      insights.push({
        id: `correlation_${Date.now()}_${Math.random()}`,
        timestamp: Date.now(),
        eventType: 'correlation_found',
        data: { correlation },
        metadata: { correlationType: 'temporal_activity' }
      })
    })

    return insights
  }

  private findGeographicCorrelations(events: AnalyticsEvent[]): Array<{
    factors: string[]
    strength: number
    sampleSize: number
  }> {
    // Simplified correlation detection
    const correlations = []

    // Example: High threat scores in specific regions
    const regionThreats = new Map<string, number[]>()

    events.forEach(event => {
      if (event.metadata.region && event.data.threatScore) {
        if (!regionThreats.has(event.metadata.region)) {
          regionThreats.set(event.metadata.region, [])
        }
        regionThreats.get(event.metadata.region)!.push(event.data.threatScore)
      }
    })

    // Find regions with consistently high threat scores
    for (const [region, scores] of regionThreats) {
      if (scores.length >= 5) {
        const avgScore = scores.reduce((a, b) => a + b, 0) / scores.length
        if (avgScore > 75) {
          correlations.push({
            factors: ['geographic_region', 'threat_score'],
            strength: Math.min(1.0, avgScore / 100),
            sampleSize: scores.length
          })
        }
      }
    }

    return correlations
  }

  private findTemporalCorrelations(events: AnalyticsEvent[]): Array<{
    factors: string[]
    strength: number
    sampleSize: number
  }> {
    // Find time-based patterns in activity
    const hourActivity = new Map<number, number>()

    events.forEach(event => {
      if (event.metadata.hour !== undefined) {
        hourActivity.set(event.metadata.hour, (hourActivity.get(event.metadata.hour) || 0) + 1)
      }
    })

    // Detect peak hours
    let maxActivity = 0
    let peakHour = 0

    for (const [hour, count] of hourActivity) {
      if (count > maxActivity) {
        maxActivity = count
        peakHour = hour
      }
    }

    if (maxActivity > 10) { // Significant peak
      return [{
        factors: ['time_of_day', 'activity_level'],
        strength: maxActivity / events.length,
        sampleSize: events.length
      }]
    }

    return []
  }
}

// Initialize analytics pipeline
const analyticsConfig: StreamingAnalyticsConfig = {
  windowSize: 300, // 5 minutes
  slideInterval: 60, // Update every minute
  minPatternSupport: 3,
  trendSensitivity: 0.2,
  anomalyThreshold: 2.0
}

const analyticsPipeline = new IPIntelligenceAnalyticsPipeline(analyticsConfig)

// Analytics API endpoints
app.get('/api/analytics/insights', async (req, res) => {
  const state = analyticsPipeline.getAnalyticsState()

  res.json({
    state,
    timestamp: new Date().toISOString()
  })
})

app.get('/api/analytics/patterns', async (req, res) => {
  const patterns = analyticsPipeline['patternDetector'].getActivePatternCount()

  res.json({
    activePatterns: patterns,
    timestamp: new Date().toISOString()
  })
})

app.get('/api/analytics/trends', async (req, res) => {
  const trends = analyticsPipeline['trendAnalyzer'].getActiveTrendCount()

  res.json({
    activeTrends: trends,
    timestamp: new Date().toISOString()
  })
})

console.log('Streaming analytics pipeline initialized')

// Real-time performance monitoring for IP intelligence systems
interface PerformanceMetric {
  name: string
  value: number
  unit: string
  timestamp: number
  tags: Record<string, string>
}

interface SystemHealth {
  status: 'healthy' | 'degraded' | 'unhealthy'
  cpuUsage: number
  memoryUsage: number
  diskUsage: number
  networkIO: { inbound: number; outbound: number }
  activeConnections: number
  errorRate: number
  responseTime: number
}

interface AlertRule {
  id: string
  name: string
  condition: {
    metric: string
    operator: '>' | '<' | '>=' | '<=' | '=='
    threshold: number
    duration: number // seconds - how long condition must be true
  }
  severity: 'low' | 'medium' | 'high' | 'critical'
  message: string
  cooldown: number // seconds between alerts
}

class PerformanceMonitor {
  private metrics: Map<string, PerformanceMetric[]> = new Map()
  private alertRules: AlertRule[] = []
  private activeAlerts: Map<string, { alertId: string; startTime: number }> = new Map()
  private subscribers: Array<(metric: PerformanceMetric) => void> = []

  constructor() {
    this.initializeDefaultAlertRules()
    this.startMetricsCollection()
  }

  // Record a performance metric
  recordMetric(name: string, value: number, unit: string, tags: Record<string, string> = {}): void {
    const metric: PerformanceMetric = {
      name,
      value,
      unit,
      timestamp: Date.now(),
      tags
    }

    if (!this.metrics.has(name)) {
      this.metrics.set(name, [])
    }

    const metricHistory = this.metrics.get(name)!
    metricHistory.push(metric)

    // Keep only last 1000 metrics per type
    if (metricHistory.length > 1000) {
      metricHistory.splice(0, metricHistory.length - 1000)
    }

    // Notify subscribers
    this.subscribers.forEach(callback => {
      try {
        callback(metric)
      } catch (error) {
        console.error('Error in performance metric subscriber:', error)
      }
    })

    // Check alert conditions
    this.checkAlertConditions(metric)
  }

  // Subscribe to real-time performance metrics
  subscribe(callback: (metric: PerformanceMetric) => void): () => void {
    this.subscribers.push(callback)

    // Return unsubscribe function
    return () => {
      const index = this.subscribers.indexOf(callback)
      if (index > -1) {
        this.subscribers.splice(index, 1)
      }
    }
  }

  // Get current system health
  getSystemHealth(): SystemHealth {
    const cpuMetric = this.getLatestMetric('cpu_usage')
    const memoryMetric = this.getLatestMetric('memory_usage')
    const diskMetric = this.getLatestMetric('disk_usage')
    const networkMetrics = this.getNetworkMetrics()
    const errorRate = this.getLatestMetric('error_rate')?.value || 0
    const responseTime = this.getLatestMetric('response_time')?.value || 0

    let status: SystemHealth['status'] = 'healthy'
    if (cpuMetric && cpuMetric.value > 80) status = 'degraded'
    if (memoryMetric && memoryMetric.value > 90) status = 'unhealthy'
    if (errorRate > 0.05) status = 'unhealthy'

    return {
      status,
      cpuUsage: cpuMetric?.value || 0,
      memoryUsage: memoryMetric?.value || 0,
      diskUsage: diskMetric?.value || 0,
      networkIO: networkMetrics,
      activeConnections: this.getLatestMetric('active_connections')?.value || 0,
      errorRate,
      responseTime
    }
  }

  // Get metrics for a specific time range
  getMetricsInRange(name: string, startTime: number, endTime: number): PerformanceMetric[] {
    const metricHistory = this.metrics.get(name) || []
    return metricHistory.filter(m => m.timestamp >= startTime && m.timestamp <= endTime)
  }

  // Get aggregated metrics (average, min, max, etc.)
  getAggregatedMetrics(name: string, timeframe: number = 300000): {
    average: number
    min: number
    max: number
    count: number
    latest: number
  } {
    const cutoff = Date.now() - timeframe
    const relevantMetrics = this.metrics.get(name)?.filter(m => m.timestamp > cutoff) || []

    if (relevantMetrics.length === 0) {
      return { average: 0, min: 0, max: 0, count: 0, latest: 0 }
    }

    const values = relevantMetrics.map(m => m.value)
    const latest = values[values.length - 1]

    return {
      average: values.reduce((a, b) => a + b, 0) / values.length,
      min: Math.min(...values),
      max: Math.max(...values),
      count: values.length,
      latest
    }
  }

  private getLatestMetric(name: string): PerformanceMetric | null {
    const metricHistory = this.metrics.get(name)
    return metricHistory && metricHistory.length > 0 ? metricHistory[metricHistory.length - 1] : null
  }

  private getNetworkMetrics(): { inbound: number; outbound: number } {
    const inbound = this.getLatestMetric('network_inbound')?.value || 0
    const outbound = this.getLatestMetric('network_outbound')?.value || 0
    return { inbound, outbound }
  }

  private checkAlertConditions(metric: PerformanceMetric): void {
    for (const rule of this.alertRules) {
      if (metric.name === rule.condition.metric) {
        const conditionMet = this.evaluateCondition(metric.value, rule.condition)

        if (conditionMet) {
          const activeAlert = this.activeAlerts.get(rule.id)

          if (!activeAlert || Date.now() - activeAlert.startTime > rule.cooldown * 1000) {
            this.triggerAlert(rule, metric)
          }
        } else {
          // Clear active alert if condition no longer met
          if (this.activeAlerts.has(rule.id)) {
            this.activeAlerts.delete(rule.id)
          }
        }
      }
    }
  }

  private evaluateCondition(value: number, condition: AlertRule['condition']): boolean {
    switch (condition.operator) {
      case '>': return value > condition.threshold
      case '<': return value < condition.threshold
      case '>=': return value >= condition.threshold
      case '<=': return value <= condition.threshold
      case '==': return Math.abs(value - condition.threshold) < 0.001
      default: return false
    }
  }

  private triggerAlert(rule: AlertRule, metric: PerformanceMetric): void {
    const alert = {
      alertId: rule.id,
      startTime: Date.now()
    }

    this.activeAlerts.set(rule.id, alert)

    // In production, this would send notifications, log to external systems, etc.
    console.warn(`🚨 ALERT [${rule.severity.toUpperCase()}]: ${rule.message}`)
    console.warn(`   Metric: ${metric.name} = ${metric.value}${metric.unit}`)
    console.warn(`   Threshold: ${rule.condition.metric} ${rule.condition.operator} ${rule.condition.threshold}`)
  }

  private initializeDefaultAlertRules(): void {
    this.alertRules = [
      {
        id: 'high_cpu',
        name: 'High CPU Usage',
        condition: {
          metric: 'cpu_usage',
          operator: '>',
          threshold: 80,
          duration: 300 // 5 minutes
        },
        severity: 'high',
        message: 'CPU usage has exceeded 80% for more than 5 minutes',
        cooldown: 1800 // 30 minutes
      },
      {
        id: 'high_memory',
        name: 'High Memory Usage',
        condition: {
          metric: 'memory_usage',
          operator: '>',
          threshold: 85,
          duration: 60 // 1 minute
        },
        severity: 'critical',
        message: 'Memory usage has exceeded 85%',
        cooldown: 600 // 10 minutes
      },
      {
        id: 'high_error_rate',
        name: 'High Error Rate',
        condition: {
          metric: 'error_rate',
          operator: '>',
          threshold: 0.05, // 5%
          duration: 180 // 3 minutes
        },
        severity: 'medium',
        message: 'Error rate has exceeded 5% for more than 3 minutes',
        cooldown: 900 // 15 minutes
      },
      {
        id: 'slow_response',
        name: 'Slow Response Time',
        condition: {
          metric: 'response_time',
          operator: '>',
          threshold: 2000, // 2 seconds
          duration: 60 // 1 minute
        },
        severity: 'medium',
        message: 'Average response time has exceeded 2 seconds',
        cooldown: 300 // 5 minutes
      }
    ]
  }

  private startMetricsCollection(): void {
    // Collect system metrics every 10 seconds
    setInterval(() => {
      this.collectSystemMetrics()
    }, 10000)

    // Collect application metrics every 30 seconds
    setInterval(() => {
      this.collectApplicationMetrics()
    }, 30000)
  }

  private collectSystemMetrics(): void {
    // In production, use system monitoring tools (e.g., os-utils, systeminformation)
    // For demo, we'll simulate metrics
    const cpuUsage = Math.random() * 100
    const memoryUsage = Math.random() * 100
    const diskUsage = Math.random() * 100

    this.recordMetric('cpu_usage', cpuUsage, '%', { source: 'system' })
    this.recordMetric('memory_usage', memoryUsage, '%', { source: 'system' })
    this.recordMetric('disk_usage', diskUsage, '%', { source: 'system' })

    // Network I/O (simulated)
    this.recordMetric('network_inbound', Math.random() * 1000, 'KB/s', { source: 'system' })
    this.recordMetric('network_outbound', Math.random() * 500, 'KB/s', { source: 'system' })
  }

  private collectApplicationMetrics(): void {
    // Collect IP intelligence specific metrics
    if (intelligenceEngine) {
      const metrics = intelligenceEngine.getMetrics()

      this.recordMetric('processed_events', metrics.processedEvents, 'count', { source: 'application' })
      this.recordMetric('average_processing_time', metrics.averageProcessingTime, 'ms', { source: 'application' })
      this.recordMetric('error_rate', metrics.errorRate, 'ratio', { source: 'application' })
      this.recordMetric('throughput', metrics.throughput, 'events/sec', { source: 'application' })
      this.recordMetric('queue_size', metrics.queueSize, 'count', { source: 'application' })
      this.recordMetric('memory_usage', metrics.memoryUsage, 'MB', { source: 'application' })
    }

    if (analyticsPipeline) {
      const state = analyticsPipeline.getAnalyticsState()

      this.recordMetric('buffer_size', state.bufferSize, 'count', { source: 'analytics' })
      this.recordMetric('window_utilization', state.windowUtilization, 'ratio', { source: 'analytics' })
      this.recordMetric('active_patterns', state.activePatterns, 'count', { source: 'analytics' })
      this.recordMetric('recent_trends', state.recentTrends, 'count', { source: 'analytics' })
      this.recordMetric('anomaly_count', state.anomalyCount, 'count', { source: 'analytics' })
    }
  }
}

// Initialize performance monitoring
const performanceMonitor = new PerformanceMonitor()

// Real-time performance dashboard API
app.get('/api/performance/health', (req, res) => {
  const health = performanceMonitor.getSystemHealth()

  res.json({
    health,
    timestamp: new Date().toISOString()
  })
})

app.get('/api/performance/metrics/:name', (req, res) => {
  const timeframe = parseInt(req.query.timeframe as string) || 300000 // 5 minutes default
  const metrics = performanceMonitor.getAggregatedMetrics(req.params.name, timeframe)

  res.json({
    metrics,
    timeframe,
    timestamp: new Date().toISOString()
  })
})

app.get('/api/performance/metrics/:name/history', (req, res) => {
  const startTime = parseInt(req.query.start as string) || Date.now() - 3600000 // 1 hour ago
  const endTime = parseInt(req.query.end as string) || Date.now()
  const metrics = performanceMonitor.getMetricsInRange(req.params.name, startTime, endTime)

  res.json({
    metrics: metrics.map(m => ({ value: m.value, timestamp: m.timestamp })),
    count: metrics.length,
    timestamp: new Date().toISOString()
  })
})

// WebSocket endpoint for real-time performance metrics
app.ws('/api/performance/stream', (ws: any) => {
  const unsubscribe = performanceMonitor.subscribe((metric) => {
    ws.send(JSON.stringify({
      type: 'performance_metric',
      data: metric,
      timestamp: new Date().toISOString()
    }))
  })

  ws.on('close', () => {
    unsubscribe()
  })
})

// Alert management endpoints
app.get('/api/performance/alerts', (req, res) => {
  const activeAlerts = Array.from(performanceMonitor['activeAlerts'].entries()).map(([ruleId, alert]) => {
    const rule = performanceMonitor['alertRules'].find(r => r.id === ruleId)
    return {
      ruleId,
      ruleName: rule?.name,
      severity: rule?.severity,
      message: rule?.message,
      startTime: alert.startTime,
      duration: Date.now() - alert.startTime
    }
  })

  res.json({
    activeAlerts,
    totalActive: activeAlerts.length,
    timestamp: new Date().toISOString()
  })
})

console.log('Performance monitoring dashboard initialized')

interface CacheStrategy {
  level: 'L1' | 'L2' | 'L3'
  ttl: number
  maxSize: number
  evictionPolicy: 'LRU' | 'LFU' | 'TTL'
}

class IntelligentCacheManager {
  private caches: Map<string, CacheStrategy> = new Map()
  
  optimizeCacheStrategy(ipPattern: string, accessFrequency: number): CacheStrategy {
    if (accessFrequency > 1000) {
      return {
        level: 'L1',
        ttl: 3600, // 1 hour
        maxSize: 1000000,
        evictionPolicy: 'LRU'
      }
    } else if (accessFrequency > 100) {
      return {
        level: 'L2', 
        ttl: 1800, // 30 minutes
        maxSize: 100000,
        evictionPolicy: 'LFU'
      }
    } else {
      return {
        level: 'L3',
        ttl: 300, // 5 minutes
        maxSize: 10000,
        evictionPolicy: 'TTL'
      }
    }
  }
}