LEDMatrix/src/plugin_system/operation_history.py

"""
Operation history and audit log.

Tracks all plugin operations and configuration changes for debugging and auditing.
"""

import json
import threading
from typing import Dict, Any, List, Optional
from datetime import datetime
from pathlib import Path
from dataclasses import dataclass, asdict

from src.config_manager_atomic import atomic_write_text
from src.logging_config import get_logger


@dataclass
class OperationRecord:
    """Record of an operation."""
    operation_id: str
    operation_type: str
    plugin_id: Optional[str]
    timestamp: datetime
    status: str
    user: Optional[str] = None
    details: Optional[Dict[str, Any]] = None
    error: Optional[str] = None
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary for serialization."""
        result = asdict(self)
        result['timestamp'] = self.timestamp.isoformat()
        return result
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'OperationRecord':
        """Create from dictionary."""
        if isinstance(data.get('timestamp'), str):
            data['timestamp'] = datetime.fromisoformat(data['timestamp'])
        return cls(**data)


class OperationHistory:
    """
    Operation history and audit log manager.
    
    Tracks all plugin operations and configuration changes.
    """
    
    def __init__(
        self,
        history_file: Optional[str] = None,
        max_records: int = 1000
    ):
        """
        Initialize operation history. The history file is read on first use,
        not here, so constructing this costs the web app's startup nothing.

        Args:
            history_file: Path to file for persisting history
            max_records: Maximum number of records to keep
        """
        self.logger = get_logger(__name__)
        self.history_file = Path(history_file) if history_file else None
        self.max_records = max_records
        self._history_loaded = False

        # In-memory history
        self._history: List[OperationRecord] = []
        self._lock = threading.RLock()

    def _ensure_loaded(self) -> None:
        """Load the history file on first use."""
        if not self._history_loaded and self.history_file and self.history_file.exists():
            self._load_history()
            self._history_loaded = True

    def record_operation(
        self,
        operation_type: str,
        plugin_id: Optional[str] = None,
        status: str = "completed",
        details: Optional[Dict[str, Any]] = None,
        error: Optional[str] = None,
        operation_id: Optional[str] = None
    ) -> str:
        """
        Record an operation in history.
        
        Args:
            operation_type: Type of operation (install, update, uninstall, etc.)
            plugin_id: Plugin identifier
            status: Operation status
            details: Optional operation details
            error: Optional error message
            operation_id: Optional operation ID
        
        Returns:
            Operation record ID
        """
        self._ensure_loaded()
        import uuid
        record_id = operation_id or str(uuid.uuid4())
        
        record = OperationRecord(
            operation_id=record_id,
            operation_type=operation_type,
            plugin_id=plugin_id,
            timestamp=datetime.now(),
            status=status,
            details=details,
            error=error
        )
        
        with self._lock:
            self._history.append(record)
            
            # Trim history if needed
            if len(self._history) > self.max_records:
                self._history = self._history[-self.max_records:]
            
            # Save to file
            self._save_history()
        
        return record_id
    
    def get_history(
        self,
        limit: int = 100,
        plugin_id: Optional[str] = None,
        operation_type: Optional[str] = None
    ) -> List[OperationRecord]:
        """
        Get operation history.
        
        Args:
            limit: Maximum number of records to return
            plugin_id: Optional filter by plugin ID
            operation_type: Optional filter by operation type
        
        Returns:
            List of operation records, sorted by timestamp (newest first)
        """
        self._ensure_loaded()
        with self._lock:
            history = self._history.copy()
        
        # Apply filters
        if plugin_id:
            history = [r for r in history if r.plugin_id == plugin_id]
        
        if operation_type:
            history = [r for r in history if r.operation_type == operation_type]
        
        # Sort by timestamp (newest first)
        history.sort(key=lambda r: r.timestamp, reverse=True)
        
        return history[:limit]
    
    def clear_history(self) -> None:
        """Clear all operation history records."""
        with self._lock:
            self._history.clear()
            self._save_history()
        self.logger.info("Operation history cleared")

    def _save_history(self) -> None:
        """Save history to file."""
        if not self.history_file:
            return
        
        try:
            # Held across the write, and written via a temp file, so two
            # threads saving at once can't interleave or truncate the file.
            with self._lock:
                history_data = [record.to_dict() for record in self._history]
            
                # Ensure directory exists
                self.history_file.parent.mkdir(parents=True, exist_ok=True)
                
                # Write to file
                atomic_write_text(self.history_file, json.dumps(history_data, indent=2))
            
        except Exception as e:
            self.logger.error(f"Error saving operation history: {e}", exc_info=True)
    
    def _load_history(self) -> None:
        """Load history from file."""
        if not self.history_file or not self.history_file.exists():
            return
        
        try:
            with open(self.history_file, 'r') as f:
                history_data = json.load(f)
            
            with self._lock:
                self._history = [
                    OperationRecord.from_dict(record_data)
                    for record_data in history_data
                ]
            
            self.logger.info(f"Loaded {len(self._history)} operation records from file")
            
        except Exception as e:
            self.logger.error(f"Error loading operation history: {e}", exc_info=True)