enum AnomalyType {
  SLOW_JOB = 'slow_job',
  HIGH_FAILURE_RATE = 'high_failure_rate',
  WORKER_UNHEALTHY = 'worker_unhealthy',
  QUEUE_BACKLOG = 'queue_backlog',
  TIMEOUT_SPIKE = 'timeout_spike',
  REPEATED_ERROR = 'repeated_error',
  MEMORY_SPIKE = 'memory_spike',
  CPU_SPIKE = 'cpu_spike',
}

enum AnomalySeverity {
  CRITICAL = 'critical',
  HIGH = 'high',
  MEDIUM = 'medium',
  LOW = 'low',
}

interface AnomalyAlert {
  id: string;
  anomalyType: AnomalyType;
  severity: AnomalySeverity;
  workerName: string;
  jobId?: string;
  message: string;
  details: Record<string, unknown>;
  detectedAt: Date;
  resolvedAt?: Date;
  resolution?: string;
}

interface RuleContext {
  workerName: string;
  status: string;
  failureRate: number;
  queueDepth: number;
  durationMs: number;
  expectedDurationMs: number;
  timeoutCount: number;
  errorRepeatCount: number;
  errorMessage: string;
  memoryMb: number;
  cpuPercent: number;
}

interface AnomalyRule {
  anomalyType: AnomalyType;
  severity: AnomalySeverity;
  description: string;
  checkFn: (ctx: RuleContext) => boolean;
  messageTemplate: string;
  cooldownSeconds: number;
}

const ANOMALY_RULES: AnomalyRule[] = [
  {
    anomalyType: AnomalyType.SLOW_JOB,
    severity: AnomalySeverity.MEDIUM,
    description: 'Job execution time exceeds expected duration',
    checkFn: (ctx) => ctx.durationMs > ctx.expectedDurationMs * 2,
    messageTemplate: 'Job took {durationMs}ms, expected {expectedDurationMs}ms',
    cooldownSeconds: 300,
  },
  {
    anomalyType: AnomalyType.HIGH_FAILURE_RATE,
    severity: AnomalySeverity.HIGH,
    description: 'Worker failure rate exceeds threshold',
    checkFn: (ctx) => ctx.failureRate > 15,
    messageTemplate: 'Failure rate {failureRate}% exceeds 15% threshold',
    cooldownSeconds: 600,
  },
  {
    anomalyType: AnomalyType.WORKER_UNHEALTHY,
    severity: AnomalySeverity.CRITICAL,
    description: 'Worker health status is unhealthy',
    checkFn: (ctx) => ctx.status === 'unhealthy',
    messageTemplate: 'Worker {workerName} is unhealthy',
    cooldownSeconds: 300,
  },
  {
    anomalyType: AnomalyType.QUEUE_BACKLOG,
    severity: AnomalySeverity.MEDIUM,
    description: 'Queue depth exceeds threshold',
    checkFn: (ctx) => ctx.queueDepth > 50,
    messageTemplate: 'Queue depth {queueDepth} exceeds threshold',
    cooldownSeconds: 300,
  },
  {
    anomalyType: AnomalyType.REPEATED_ERROR,
    severity: AnomalySeverity.HIGH,
    description: 'Same error repeated multiple times',
    checkFn: (ctx) => ctx.errorRepeatCount > 5,
    messageTemplate: 'Error "{errorMessage}" repeated {errorRepeatCount} times',
    cooldownSeconds: 900,
  },
  {
    anomalyType: AnomalyType.MEMORY_SPIKE,
    severity: AnomalySeverity.HIGH,
    description: 'Memory usage exceeds threshold',
    checkFn: (ctx) => ctx.memoryMb > 1024,
    messageTemplate: 'Memory usage {memoryMb}MB exceeds 1GB threshold',
    cooldownSeconds: 300,
  },
];

class AnomalyDetector {
  private alerts = new Map<string, AnomalyAlert>();
  private cooldowns = new Map<string, Date>();
  private errorCounts = new Map<string, Map<string, number>>();
  private timeoutCounts = new Map<string, number>();
  private alertIdCounter = 0;

  checkWorkerHealth(
    workerName: string,
    health: {
      status: string;
      jobsProcessed: number;
      jobsFailed: number;
      queueDepth: number;
      lastDurationMs: number;
      expectedDurationMs: number;
      memoryMb: number;
      cpuPercent: number;
    }
  ): AnomalyAlert[] {
    const detected: AnomalyAlert[] = [];
    
    const failureRate = health.jobsProcessed > 0
      ? (health.jobsFailed / health.jobsProcessed) * 100
      : 0;

    const ctx: RuleContext = {
      workerName,
      status: health.status,
      failureRate,
      queueDepth: health.queueDepth,
      durationMs: health.lastDurationMs,
      expectedDurationMs: health.expectedDurationMs,
      timeoutCount: this.timeoutCounts.get(workerName) || 0,
      errorRepeatCount: 0,
      errorMessage: '',
      memoryMb: health.memoryMb,
      cpuPercent: health.cpuPercent,
    };

    for (const rule of ANOMALY_RULES) {
      if (this.isOnCooldown(workerName, rule.anomalyType)) continue;

      if (rule.checkFn(ctx)) {
        const alert = this.createAlert(workerName, rule, ctx);
        detected.push(alert);
        this.setCooldown(workerName, rule.anomalyType, rule.cooldownSeconds);
      }
    }

    return detected;
  }

  checkJobExecution(
    workerName: string,
    jobId: string,
    durationMs: number,
    expectedDurationMs: number,
    success: boolean,
    error?: string
  ): AnomalyAlert[] {
    const detected: AnomalyAlert[] = [];

    if (!success && error) {
      this.trackError(workerName, error);
    }

    // Check slow job
    if (durationMs > expectedDurationMs * 2) {
      if (!this.isOnCooldown(workerName, AnomalyType.SLOW_JOB)) {
        const rule = ANOMALY_RULES[0];
        const alert = this.createAlert(workerName, rule, {
          durationMs,
          expectedDurationMs,
        } as RuleContext);
        alert.jobId = jobId;
        detected.push(alert);
        this.setCooldown(workerName, AnomalyType.SLOW_JOB, 300);
      }
    }

    // Check repeated errors
    if (error) {
      const errorCounts = this.errorCounts.get(workerName);
      const count = errorCounts?.get(error.slice(0, 200)) || 0;
      
      if (count > 5 && !this.isOnCooldown(workerName, AnomalyType.REPEATED_ERROR)) {
        const rule = ANOMALY_RULES.find(r => r.anomalyType === AnomalyType.REPEATED_ERROR)!;
        const alert = this.createAlert(workerName, rule, {
          errorMessage: error.slice(0, 100),
          errorRepeatCount: count,
        } as RuleContext);
        detected.push(alert);
        this.setCooldown(workerName, AnomalyType.REPEATED_ERROR, 900);
      }
    }

    return detected;
  }

  resolveAnomaly(alertId: string, resolution: string): boolean {
    const alert = this.alerts.get(alertId);
    if (!alert || alert.resolvedAt) return false;

    alert.resolvedAt = new Date();
    alert.resolution = resolution;
    return true;
  }

  getActiveAnomalies(): AnomalyAlert[] {
    return Array.from(this.alerts.values())
      .filter(a => !a.resolvedAt)
      .sort((a, b) => {
        const order = { critical: 0, high: 1, medium: 2, low: 3 };
        return order[a.severity] - order[b.severity];
      });
  }

  private trackError(workerName: string, error: string): void {
    if (!this.errorCounts.has(workerName)) {
      this.errorCounts.set(workerName, new Map());
    }
    const counts = this.errorCounts.get(workerName)!;
    const key = error.slice(0, 200);
    counts.set(key, (counts.get(key) || 0) + 1);
  }

  private createAlert(workerName: string, rule: AnomalyRule, ctx: Partial<RuleContext>): AnomalyAlert {
    const id = `anomaly_${++this.alertIdCounter}_${Date.now()}`;
    
    let message = rule.messageTemplate;
    for (const [key, value] of Object.entries(ctx)) {
      message = message.replace(`{${key}}`, String(value));
    }

    const alert: AnomalyAlert = {
      id,
      anomalyType: rule.anomalyType,
      severity: rule.severity,
      workerName,
      message,
      details: ctx as Record<string, unknown>,
      detectedAt: new Date(),
    };

    this.alerts.set(id, alert);
    return alert;
  }

  private isOnCooldown(workerName: string, anomalyType: AnomalyType): boolean {
    const key = `${workerName}:${anomalyType}`;
    const cooldownEnd = this.cooldowns.get(key);
    return cooldownEnd !== undefined && cooldownEnd > new Date();
  }

  private setCooldown(workerName: string, anomalyType: AnomalyType, seconds: number): void {
    const key = `${workerName}:${anomalyType}`;
    this.cooldowns.set(key, new Date(Date.now() + seconds * 1000));
  }
}

// Singleton
let detector: AnomalyDetector | null = null;

export function getAnomalyDetector(): AnomalyDetector {
  if (!detector) {
    detector = new AnomalyDetector();
  }
  return detector;
}

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, List, Optional, Callable
from enum import Enum


class AnomalyType(str, Enum):
    SLOW_JOB = "slow_job"
    HIGH_FAILURE_RATE = "high_failure_rate"
    WORKER_UNHEALTHY = "worker_unhealthy"
    QUEUE_BACKLOG = "queue_backlog"
    TIMEOUT_SPIKE = "timeout_spike"
    REPEATED_ERROR = "repeated_error"
    MEMORY_SPIKE = "memory_spike"


class AnomalySeverity(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


@dataclass
class AnomalyAlert:
    id: str
    anomaly_type: AnomalyType
    severity: AnomalySeverity
    worker_name: str
    message: str
    details: Dict
    detected_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    job_id: Optional[str] = None
    resolved_at: Optional[datetime] = None
    resolution: Optional[str] = None


@dataclass
class RuleContext:
    worker_name: str
    status: str = "healthy"
    failure_rate: float = 0.0
    queue_depth: int = 0
    duration_ms: float = 0.0
    expected_duration_ms: float = 0.0
    timeout_count: int = 0
    error_repeat_count: int = 0
    error_message: str = ""
    memory_mb: float = 0.0
    cpu_percent: float = 0.0


@dataclass
class AnomalyRule:
    anomaly_type: AnomalyType
    severity: AnomalySeverity
    description: str
    check_fn: Callable[[RuleContext], bool]
    message_template: str
    cooldown_seconds: int


ANOMALY_RULES: List[AnomalyRule] = [
    AnomalyRule(
        anomaly_type=AnomalyType.SLOW_JOB,
        severity=AnomalySeverity.MEDIUM,
        description="Job execution time exceeds expected duration",
        check_fn=lambda ctx: ctx.duration_ms > ctx.expected_duration_ms * 2,
        message_template="Job took {duration_ms}ms, expected {expected_duration_ms}ms",
        cooldown_seconds=300,
    ),
    AnomalyRule(
        anomaly_type=AnomalyType.HIGH_FAILURE_RATE,
        severity=AnomalySeverity.HIGH,
        description="Worker failure rate exceeds threshold",
        check_fn=lambda ctx: ctx.failure_rate > 15,
        message_template="Failure rate {failure_rate}% exceeds 15% threshold",
        cooldown_seconds=600,
    ),
    AnomalyRule(
        anomaly_type=AnomalyType.WORKER_UNHEALTHY,
        severity=AnomalySeverity.CRITICAL,
        description="Worker health status is unhealthy",
        check_fn=lambda ctx: ctx.status == "unhealthy",
        message_template="Worker {worker_name} is unhealthy",
        cooldown_seconds=300,
    ),
    AnomalyRule(
        anomaly_type=AnomalyType.REPEATED_ERROR,
        severity=AnomalySeverity.HIGH,
        description="Same error repeated multiple times",
        check_fn=lambda ctx: ctx.error_repeat_count > 5,
        message_template='Error "{error_message}" repeated {error_repeat_count} times',
        cooldown_seconds=900,
    ),
]


class AnomalyDetector:
    def __init__(self):
        self._alerts: Dict[str, AnomalyAlert] = {}
        self._cooldowns: Dict[str, datetime] = {}
        self._error_counts: Dict[str, Dict[str, int]] = {}
        self._timeout_counts: Dict[str, int] = {}
        self._alert_counter = 0
    
    def check_worker_health(
        self,
        worker_name: str,
        status: str,
        jobs_processed: int,
        jobs_failed: int,
        queue_depth: int,
        last_duration_ms: float,
        expected_duration_ms: float,
        memory_mb: float = 0,
        cpu_percent: float = 0,
    ) -> List[AnomalyAlert]:
        detected: List[AnomalyAlert] = []
        
        failure_rate = (jobs_failed / jobs_processed * 100) if jobs_processed > 0 else 0
        
        ctx = RuleContext(
            worker_name=worker_name,
            status=status,
            failure_rate=failure_rate,
            queue_depth=queue_depth,
            duration_ms=last_duration_ms,
            expected_duration_ms=expected_duration_ms,
            timeout_count=self._timeout_counts.get(worker_name, 0),
            memory_mb=memory_mb,
            cpu_percent=cpu_percent,
        )
        
        for rule in ANOMALY_RULES:
            if self._is_on_cooldown(worker_name, rule.anomaly_type):
                continue
            
            if rule.check_fn(ctx):
                alert = self._create_alert(worker_name, rule, ctx)
                detected.append(alert)
                self._set_cooldown(worker_name, rule.anomaly_type, rule.cooldown_seconds)
        
        return detected
    
    def check_job_execution(
        self,
        worker_name: str,
        job_id: str,
        duration_ms: float,
        expected_duration_ms: float,
        success: bool,
        error: Optional[str] = None,
    ) -> List[AnomalyAlert]:
        detected: List[AnomalyAlert] = []
        
        if not success and error:
            self._track_error(worker_name, error)
        
        # Check slow job
        if duration_ms > expected_duration_ms * 2:
            if not self._is_on_cooldown(worker_name, AnomalyType.SLOW_JOB):
                rule = ANOMALY_RULES[0]
                ctx = RuleContext(
                    worker_name=worker_name,
                    duration_ms=duration_ms,
                    expected_duration_ms=expected_duration_ms,
                )
                alert = self._create_alert(worker_name, rule, ctx)
                alert.job_id = job_id
                detected.append(alert)
                self._set_cooldown(worker_name, AnomalyType.SLOW_JOB, 300)
        
        # Check repeated errors
        if error:
            error_counts = self._error_counts.get(worker_name, {})
            count = error_counts.get(error[:200], 0)
            
            if count > 5 and not self._is_on_cooldown(worker_name, AnomalyType.REPEATED_ERROR):
                rule = next(r for r in ANOMALY_RULES if r.anomaly_type == AnomalyType.REPEATED_ERROR)
                ctx = RuleContext(
                    worker_name=worker_name,
                    error_message=error[:100],
                    error_repeat_count=count,
                )
                alert = self._create_alert(worker_name, rule, ctx)
                detected.append(alert)
                self._set_cooldown(worker_name, AnomalyType.REPEATED_ERROR, 900)
        
        return detected
    
    def resolve_anomaly(self, alert_id: str, resolution: str) -> bool:
        alert = self._alerts.get(alert_id)
        if not alert or alert.resolved_at:
            return False
        
        alert.resolved_at = datetime.now(timezone.utc)
        alert.resolution = resolution
        return True
    
    def get_active_anomalies(self) -> List[AnomalyAlert]:
        severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
        return sorted(
            [a for a in self._alerts.values() if not a.resolved_at],
            key=lambda a: severity_order[a.severity.value]
        )
    
    def _track_error(self, worker_name: str, error: str) -> None:
        if worker_name not in self._error_counts:
            self._error_counts[worker_name] = {}
        key = error[:200]
        self._error_counts[worker_name][key] = self._error_counts[worker_name].get(key, 0) + 1
    
    def _create_alert(self, worker_name: str, rule: AnomalyRule, ctx: RuleContext) -> AnomalyAlert:
        self._alert_counter += 1
        alert_id = f"anomaly_{self._alert_counter}_{int(datetime.now().timestamp() * 1000)}"
        
        message = rule.message_template
        for key, value in ctx.__dict__.items():
            message = message.replace(f"{{{key}}}", str(value))
        
        alert = AnomalyAlert(
            id=alert_id,
            anomaly_type=rule.anomaly_type,
            severity=rule.severity,
            worker_name=worker_name,
            message=message,
            details=ctx.__dict__,
        )
        
        self._alerts[alert_id] = alert
        return alert
    
    def _is_on_cooldown(self, worker_name: str, anomaly_type: AnomalyType) -> bool:
        key = f"{worker_name}:{anomaly_type.value}"
        cooldown_end = self._cooldowns.get(key)
        return cooldown_end is not None and cooldown_end > datetime.now(timezone.utc)
    
    def _set_cooldown(self, worker_name: str, anomaly_type: AnomalyType, seconds: int) -> None:
        from datetime import timedelta
        key = f"{worker_name}:{anomaly_type.value}"
        self._cooldowns[key] = datetime.now(timezone.utc) + timedelta(seconds=seconds)