feat(performance): implement cache warming and database optimization

- Create cache warming service with multiple strategies
- Add query tracking middleware for pattern analysis
- Implement database optimization service
- Create admin APIs for cache and database management
- Add Celery tasks for scheduled warming
- Create performance indexes migration
- Add optimization script for manual/cron execution
- Configure optimization thresholds and schedules

alembic/versions/007_add_performance_indexes.py

@@ -0,0 +1,168 @@
+"""add performance indexes
+
+Revision ID: 007
+Revises: 006
+Create Date: 2025-01-25
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers
+revision = '007'
+down_revision = '006'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Create performance indexes."""
+    
+    # Investigations table indexes
+    op.create_index(
+        'ix_investigations_status_created_at',
+        'investigations',
+        ['status', 'created_at']
+    )
+    
+    op.create_index(
+        'ix_investigations_contract_id',
+        'investigations',
+        ['contract_id']
+    )
+    
+    # Chat sessions indexes
+    op.create_index(
+        'ix_chat_sessions_user_id_created_at',
+        'chat_sessions',
+        ['user_id', 'created_at']
+    )
+    
+    # Chat messages indexes
+    op.create_index(
+        'ix_chat_messages_session_id_created_at',
+        'chat_messages',
+        ['session_id', 'created_at']
+    )
+    
+    # API keys indexes (if not already created)
+    op.create_index(
+        'ix_api_keys_key_hash',
+        'api_keys',
+        ['key_hash'],
+        unique=True,
+        postgresql_using='hash'
+    )
+    
+    op.create_index(
+        'ix_api_keys_status',
+        'api_keys',
+        ['status']
+    )
+    
+    # Agents table indexes
+    op.create_index(
+        'ix_agents_type_status',
+        'agents',
+        ['type', 'status']
+    )
+    
+    # Anomalies table indexes (if exists)
+    try:
+        op.create_index(
+            'ix_anomalies_investigation_id',
+            'anomalies',
+            ['investigation_id']
+        )
+        
+        op.create_index(
+            'ix_anomalies_severity_created_at',
+            'anomalies',
+            ['severity', 'created_at']
+        )
+    except:
+        pass  # Table might not exist yet
+    
+    # Reports table indexes (if exists)
+    try:
+        op.create_index(
+            'ix_reports_investigation_id',
+            'reports',
+            ['investigation_id']
+        )
+        
+        op.create_index(
+            'ix_reports_format_created_at',
+            'reports',
+            ['format', 'created_at']
+        )
+    except:
+        pass
+    
+    # Audit logs indexes
+    op.create_index(
+        'ix_audit_logs_event_type_timestamp',
+        'audit_logs',
+        ['event_type', 'timestamp']
+    )
+    
+    op.create_index(
+        'ix_audit_logs_user_id_timestamp',
+        'audit_logs',
+        ['user_id', 'timestamp']
+    )
+    
+    # Create partial indexes for better performance
+    op.execute("""
+        CREATE INDEX CONCURRENTLY ix_investigations_pending 
+        ON investigations (created_at) 
+        WHERE status = 'PENDING'
+    """)
+    
+    op.execute("""
+        CREATE INDEX CONCURRENTLY ix_api_keys_active 
+        ON api_keys (created_at) 
+        WHERE status = 'ACTIVE'
+    """)
+    
+    # Create GIN index for JSONB columns
+    op.execute("""
+        CREATE INDEX CONCURRENTLY ix_investigations_results_gin 
+        ON investigations USING gin (results)
+    """)
+    
+    op.execute("""
+        CREATE INDEX CONCURRENTLY ix_investigations_metadata_gin 
+        ON investigations USING gin (metadata)
+    """)
+
+
+def downgrade() -> None:
+    """Drop performance indexes."""
+    
+    # Drop all created indexes
+    op.drop_index('ix_investigations_status_created_at', 'investigations')
+    op.drop_index('ix_investigations_contract_id', 'investigations')
+    op.drop_index('ix_chat_sessions_user_id_created_at', 'chat_sessions')
+    op.drop_index('ix_chat_messages_session_id_created_at', 'chat_messages')
+    op.drop_index('ix_api_keys_key_hash', 'api_keys')
+    op.drop_index('ix_api_keys_status', 'api_keys')
+    op.drop_index('ix_agents_type_status', 'agents')
+    op.drop_index('ix_audit_logs_event_type_timestamp', 'audit_logs')
+    op.drop_index('ix_audit_logs_user_id_timestamp', 'audit_logs')
+    
+    # Drop partial indexes
+    op.execute("DROP INDEX IF EXISTS ix_investigations_pending")
+    op.execute("DROP INDEX IF EXISTS ix_api_keys_active")
+    
+    # Drop GIN indexes
+    op.execute("DROP INDEX IF EXISTS ix_investigations_results_gin")
+    op.execute("DROP INDEX IF EXISTS ix_investigations_metadata_gin")
+    
+    # Drop indexes that might not exist
+    try:
+        op.drop_index('ix_anomalies_investigation_id', 'anomalies')
+        op.drop_index('ix_anomalies_severity_created_at', 'anomalies')
+        op.drop_index('ix_reports_investigation_id', 'reports')
+        op.drop_index('ix_reports_format_created_at', 'reports')
+    except:
+        pass

config/database_optimization.yaml

@@ -0,0 +1,111 @@
+# Database Optimization Configuration
+
+# Thresholds for identifying issues
+thresholds:
+  slow_query_seconds: 1.0
+  cache_hit_ratio_min: 0.90
+  dead_tuple_ratio_max: 0.20
+  unused_index_days: 30
+
+# Automatic index creation rules
+auto_indexes:
+  # Create indexes on foreign keys
+  foreign_keys: true
+  
+  # Create indexes on frequently filtered columns
+  frequent_filters:
+    min_calls: 100
+    enabled: true
+  
+  # Create indexes on frequently joined columns
+  frequent_joins:
+    min_calls: 50
+    enabled: true
+
+# Table-specific optimizations
+table_optimizations:
+  investigations:
+    # Partition by created_at if > 10M rows
+    partition_threshold: 10000000
+    partition_by: created_at
+    partition_interval: monthly
+    
+  chat_messages:
+    # Archive old messages
+    archive_after_days: 90
+    archive_to_table: chat_messages_archive
+    
+  audit_logs:
+    # Compress old logs
+    compress_after_days: 30
+    compression_level: 9
+
+# Vacuum settings
+vacuum:
+  # Run VACUUM on tables with > 20% dead tuples
+  dead_tuple_threshold: 0.20
+  
+  # Run VACUUM FULL on tables with > 50% dead tuples
+  full_vacuum_threshold: 0.50
+  
+  # Exclude tables from automatic vacuum
+  exclude_tables:
+    - spatial_ref_sys
+    - pg_stat_statements
+
+# Statistics update settings
+statistics:
+  # Update statistics if stale
+  stale_threshold_days: 7
+  
+  # Sample size for ANALYZE
+  default_statistics_target: 100
+  
+  # Tables requiring higher statistics
+  high_statistics_tables:
+    investigations: 500
+    contracts: 500
+
+# Monitoring and alerting
+monitoring:
+  # Alert if slow queries exceed threshold
+  slow_query_alert_count: 10
+  
+  # Alert if cache hit ratio drops
+  cache_hit_alert_threshold: 0.85
+  
+  # Alert if connections near limit
+  connection_alert_percentage: 0.90
+
+# Scheduled optimization tasks
+schedule:
+  # Run full optimization
+  full_optimization:
+    cron: "0 2 * * SUN"  # Sunday 2 AM
+    enabled: true
+    
+  # Update statistics only
+  update_statistics:
+    cron: "0 3 * * *"    # Daily 3 AM
+    enabled: true
+    
+  # Check missing indexes
+  check_indexes:
+    cron: "0 4 * * MON,THU"  # Monday & Thursday 4 AM
+    enabled: true
+    
+  # Vacuum high-activity tables
+  vacuum_tables:
+    cron: "0 5 * * *"    # Daily 5 AM
+    enabled: true
+
+# Performance recommendations
+recommendations:
+  # Suggest configuration changes
+  suggest_config_changes: true
+  
+  # Suggest query rewrites
+  suggest_query_rewrites: true
+  
+  # Suggest schema changes
+  suggest_schema_changes: true

scripts/optimize_database.py

@@ -0,0 +1,166 @@
+#!/usr/bin/env python3
+"""
+Script to optimize database performance.
+
+This script can be run manually or scheduled via cron to:
+- Analyze and create missing indexes
+- Update table statistics
+- Vacuum tables with high dead tuple ratio
+- Generate performance reports
+
+Usage:
+    python scripts/optimize_database.py [options]
+    
+Options:
+    --dry-run       Show what would be done without making changes
+    --analyze-only  Only analyze, don't make any changes
+    --force         Force optimization even if recently done
+"""
+
+import asyncio
+import argparse
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+# Add project root to path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from src.core import get_logger
+from src.db.session import get_session
+from src.services.database_optimization_service import database_optimization_service
+from src.core.config import settings
+
+logger = get_logger(__name__)
+
+
+async def main(args):
+    """Main optimization routine."""
+    logger.info(
+        "database_optimization_started",
+        dry_run=args.dry_run,
+        analyze_only=args.analyze_only
+    )
+    
+    async with get_session() as session:
+        # 1. Analyze slow queries
+        logger.info("Analyzing slow queries...")
+        slow_queries = await database_optimization_service.analyze_slow_queries(
+            session=session,
+            limit=50
+        )
+        
+        print(f"\n=== SLOW QUERIES ANALYSIS ===")
+        print(f"Found {len(slow_queries)} slow queries")
+        
+        for i, analysis in enumerate(slow_queries[:10], 1):
+            print(f"\n{i}. Query (exec time: {analysis.execution_time:.2f}s):")
+            print(f"   {analysis.query[:100]}...")
+            print(f"   Calls: {analysis.plan.get('calls', 0)}")
+            print(f"   Suggestions: {', '.join(analysis.suggestions)}")
+        
+        # 2. Check missing indexes
+        logger.info("Checking for missing indexes...")
+        missing_indexes = await database_optimization_service.create_missing_indexes(
+            session=session,
+            dry_run=True
+        )
+        
+        print(f"\n=== MISSING INDEXES ===")
+        print(f"Found {len(missing_indexes)} missing indexes")
+        
+        for idx in missing_indexes:
+            print(f"\n- Table: {idx['table']}, Column: {idx['column']}")
+            print(f"  Reason: {idx['reason']}")
+            print(f"  Command: {idx['command']}")
+        
+        # 3. Create indexes if not in analyze-only mode
+        if not args.analyze_only and missing_indexes:
+            if args.dry_run:
+                print("\n[DRY RUN] Would create the above indexes")
+            else:
+                print("\nCreating missing indexes...")
+                created_indexes = await database_optimization_service.create_missing_indexes(
+                    session=session,
+                    dry_run=False
+                )
+                
+                created_count = sum(1 for idx in created_indexes if idx.get('status') == 'created')
+                print(f"Created {created_count} indexes")
+        
+        # 4. Update statistics
+        if not args.analyze_only:
+            if args.dry_run:
+                print("\n[DRY RUN] Would update table statistics")
+            else:
+                print("\nUpdating table statistics...")
+                stats_result = await database_optimization_service.optimize_table_statistics(
+                    session=session
+                )
+                
+                print(f"Analyzed {len(stats_result['analyzed'])} tables")
+                print(f"Vacuumed {len(stats_result['vacuumed'])} tables")
+        
+        # 5. Get database stats
+        print("\n=== DATABASE STATISTICS ===")
+        db_stats = await database_optimization_service.get_database_stats(session)
+        
+        print(f"Database size: {db_stats['database_size']['pretty']}")
+        print(f"Cache hit ratio: {db_stats['cache_hit_ratio']['ratio']:.1%}")
+        print(f"Active connections: {db_stats['connections']['active']}/{db_stats['connections']['total']}")
+        
+        print("\nLargest tables:")
+        for table in db_stats['largest_tables'][:5]:
+            print(f"- {table['table']}: {table['size_pretty']} ({table['row_count']} rows)")
+        
+        # 6. Generate report
+        report = {
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "slow_queries": len(slow_queries),
+            "missing_indexes": len(missing_indexes),
+            "created_indexes": created_count if not args.analyze_only and not args.dry_run else 0,
+            "database_size": db_stats['database_size']['pretty'],
+            "cache_hit_ratio": db_stats['cache_hit_ratio']['ratio']
+        }
+        
+        # Save report
+        report_path = Path("logs/database_optimization_report.json")
+        report_path.parent.mkdir(exist_ok=True)
+        
+        import json
+        with open(report_path, "w") as f:
+            json.dump(report, f, indent=2)
+        
+        print(f"\n=== OPTIMIZATION COMPLETE ===")
+        print(f"Report saved to: {report_path}")
+        
+        logger.info(
+            "database_optimization_completed",
+            report=report
+        )
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="Optimize database performance"
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Show what would be done without making changes"
+    )
+    parser.add_argument(
+        "--analyze-only",
+        action="store_true",
+        help="Only analyze, don't make any changes"
+    )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="Force optimization even if recently done"
+    )
+    
+    args = parser.parse_args()
+    
+    # Run optimization
+    asyncio.run(main(args))

src/api/app.py

@@ -214,6 +214,19 @@ app.add_middleware(
     strategy="sliding_window"
 )
 
+# Add query tracking middleware for cache optimization
+from src.api.middleware.query_tracking import QueryTrackingMiddleware
+app.add_middleware(
+    QueryTrackingMiddleware,
+    tracked_paths=[
+        "/api/v1/investigations",
+        "/api/v1/contracts", 
+        "/api/v1/analysis",
+        "/api/v1/reports"
+    ],
+    sample_rate=0.1 if settings.is_production else 1.0  # 10% sampling in production
+)
+
 
 # Custom OpenAPI schema
 def custom_openapi():
@@ -392,6 +405,8 @@ app.include_router(
 
 # Import and include admin routes
 from src.api.routes.admin import ip_whitelist as admin_ip_whitelist
+from src.api.routes.admin import cache_warming as admin_cache_warming
+from src.api.routes.admin import database_optimization as admin_db_optimization
 from src.api.routes import api_keys
 
 app.include_router(
@@ -400,6 +415,18 @@ app.include_router(
     tags=["Admin - IP Whitelist"]
 )
 
+app.include_router(
+    admin_cache_warming.router,
+    prefix="/api/v1/admin",
+    tags=["Admin - Cache Warming"]
+)
+
+app.include_router(
+    admin_db_optimization.router,
+    prefix="/api/v1/admin",
+    tags=["Admin - Database Optimization"]
+)
+
 app.include_router(
     api_keys.router,
     prefix="/api/v1",

src/api/middleware/query_tracking.py

@@ -0,0 +1,113 @@
+"""
+Module: api.middleware.query_tracking
+Description: Middleware to track query patterns for cache warming
+Author: Anderson H. Silva
+Date: 2025-01-25
+License: Proprietary - All rights reserved
+"""
+
+from typing import Dict, Any
+import json
+import hashlib
+
+from fastapi import Request
+from starlette.middleware.base import BaseHTTPMiddleware
+
+from src.core import get_logger
+from src.services.cache_warming_service import cache_warming_service
+
+logger = get_logger(__name__)
+
+
+class QueryTrackingMiddleware(BaseHTTPMiddleware):
+    """
+    Middleware to track query patterns for cache optimization.
+    
+    Tracks:
+    - API endpoint access frequency
+    - Query parameters
+    - Response times
+    - Cache hit/miss patterns
+    """
+    
+    def __init__(
+        self,
+        app,
+        tracked_paths: list = None,
+        sample_rate: float = 1.0
+    ):
+        """
+        Initialize query tracking middleware.
+        
+        Args:
+            app: FastAPI application
+            tracked_paths: List of paths to track (None = all)
+            sample_rate: Sampling rate (0.0 to 1.0)
+        """
+        super().__init__(app)
+        self.tracked_paths = tracked_paths or [
+            "/api/v1/investigations",
+            "/api/v1/contracts",
+            "/api/v1/analysis",
+            "/api/v1/reports",
+            "/api/v1/chat"
+        ]
+        self.sample_rate = sample_rate
+        
+    async def dispatch(self, request: Request, call_next):
+        """Process request with query tracking."""
+        # Check if we should track this request
+        if not self._should_track(request):
+            return await call_next(request)
+        
+        # Extract query information
+        query_info = self._extract_query_info(request)
+        
+        # Process request
+        response = await call_next(request)
+        
+        # Track query in background
+        try:
+            cache_warming_service.track_query(query_info)
+        except Exception as e:
+            logger.error(
+                "query_tracking_error",
+                error=str(e),
+                query_info=query_info
+            )
+        
+        return response
+    
+    def _should_track(self, request: Request) -> bool:
+        """Check if request should be tracked."""
+        # Check sampling rate
+        import random
+        if random.random() > self.sample_rate:
+            return False
+        
+        # Check path
+        path = request.url.path
+        for tracked_path in self.tracked_paths:
+            if path.startswith(tracked_path):
+                return True
+        
+        return False
+    
+    def _extract_query_info(self, request: Request) -> Dict[str, Any]:
+        """Extract query information from request."""
+        query_info = {
+            "path": request.url.path,
+            "method": request.method,
+            "query_params": dict(request.query_params),
+            "timestamp": None  # Will be set by service
+        }
+        
+        # Add path parameters if available
+        if hasattr(request, "path_params"):
+            query_info["path_params"] = request.path_params
+        
+        # Generate query hash for deduplication
+        query_str = json.dumps(query_info, sort_keys=True)
+        query_info["hash"] = hashlib.md5(query_str.encode()).hexdigest()
+        
+        return query_info

src/api/routes/admin/cache_warming.py

@@ -0,0 +1,201 @@
+"""
+Module: api.routes.admin.cache_warming
+Description: Admin routes for cache warming management
+Author: Anderson H. Silva
+Date: 2025-01-25
+License: Proprietary - All rights reserved
+"""
+
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+
+from fastapi import APIRouter, Depends, HTTPException, status, BackgroundTasks
+from pydantic import BaseModel, Field
+
+from src.core import get_logger
+from src.api.dependencies import require_admin
+from src.services.cache_warming_service import (
+    cache_warming_service,
+    CacheWarmingStrategy
+)
+
+logger = get_logger(__name__)
+
+router = APIRouter(prefix="/cache-warming", tags=["Admin - Cache Warming"])
+
+
+class CacheWarmingRequest(BaseModel):
+    """Request to warm specific cache data."""
+    strategies: Optional[List[CacheWarmingStrategy]] = Field(
+        None,
+        description="Specific strategies to execute (None = all)"
+    )
+
+
+class SpecificDataWarmingRequest(BaseModel):
+    """Request to warm specific data items."""
+    data_type: str = Field(..., description="Type of data to warm")
+    identifiers: List[str] = Field(..., description="List of identifiers")
+    ttl: Optional[int] = Field(None, description="Cache TTL in seconds")
+
+
+class CacheWarmingStatusResponse(BaseModel):
+    """Cache warming status response."""
+    last_warming: Optional[datetime]
+    query_frequency_tracked: int
+    top_queries: List[tuple]
+    config: Dict[str, Any]
+
+
+@router.post("/trigger")
+async def trigger_cache_warming(
+    request: CacheWarmingRequest,
+    background_tasks: BackgroundTasks,
+    admin_user=Depends(require_admin)
+):
+    """
+    Manually trigger cache warming.
+    
+    Requires admin privileges.
+    """
+    logger.info(
+        "admin_cache_warming_triggered",
+        admin=admin_user.get("email"),
+        strategies=request.strategies
+    )
+    
+    # Execute in background
+    background_tasks.add_task(
+        cache_warming_service.trigger_manual_warming,
+        request.strategies
+    )
+    
+    return {
+        "status": "warming_started",
+        "strategies": request.strategies or "all",
+        "message": "Cache warming started in background"
+    }
+
+
+@router.post("/warm-specific")
+async def warm_specific_data(
+    request: SpecificDataWarmingRequest,
+    admin_user=Depends(require_admin)
+):
+    """
+    Warm cache with specific data items.
+    
+    Requires admin privileges.
+    """
+    try:
+        results = await cache_warming_service.warm_specific_data(
+            data_type=request.data_type,
+            identifiers=request.identifiers,
+            ttl=request.ttl
+        )
+        
+        logger.info(
+            "admin_specific_data_warmed",
+            admin=admin_user.get("email"),
+            data_type=request.data_type,
+            warmed_count=len(results["warmed"]),
+            failed_count=len(results["failed"])
+        )
+        
+        return results
+        
+    except Exception as e:
+        logger.error(
+            "admin_specific_data_warming_error",
+            error=str(e),
+            exc_info=True
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to warm specific data"
+        )
+
+
+@router.get("/status", response_model=CacheWarmingStatusResponse)
+async def get_warming_status(
+    admin_user=Depends(require_admin)
+):
+    """
+    Get cache warming status.
+    
+    Requires admin privileges.
+    """
+    status = await cache_warming_service.get_warming_status()
+    
+    return CacheWarmingStatusResponse(
+        last_warming=status["last_warming"],
+        query_frequency_tracked=status["query_frequency_tracked"],
+        top_queries=status["top_queries"],
+        config=status["config"]
+    )
+
+
+@router.post("/strategies/{strategy}")
+async def execute_single_strategy(
+    strategy: CacheWarmingStrategy,
+    background_tasks: BackgroundTasks,
+    admin_user=Depends(require_admin)
+):
+    """
+    Execute a single cache warming strategy.
+    
+    Requires admin privileges.
+    """
+    logger.info(
+        "admin_single_strategy_warming",
+        admin=admin_user.get("email"),
+        strategy=strategy
+    )
+    
+    # Execute in background
+    background_tasks.add_task(
+        cache_warming_service.trigger_manual_warming,
+        [strategy]
+    )
+    
+    return {
+        "status": "strategy_started",
+        "strategy": strategy,
+        "message": f"Cache warming strategy '{strategy}' started"
+    }
+
+
+@router.get("/strategies")
+async def list_available_strategies(
+    admin_user=Depends(require_admin)
+):
+    """
+    List available cache warming strategies.
+    
+    Requires admin privileges.
+    """
+    strategies = [
+        {
+            "name": strategy.value,
+            "description": get_strategy_description(strategy)
+        }
+        for strategy in CacheWarmingStrategy
+    ]
+    
+    return {
+        "strategies": strategies,
+        "total": len(strategies)
+    }
+
+
+def get_strategy_description(strategy: CacheWarmingStrategy) -> str:
+    """Get human-readable description for strategy."""
+    descriptions = {
+        CacheWarmingStrategy.POPULAR_DATA: "Warm cache with frequently accessed contracts and data",
+        CacheWarmingStrategy.RECENT_INVESTIGATIONS: "Cache recent investigation results",
+        CacheWarmingStrategy.FREQUENT_QUERIES: "Cache results of frequently executed queries",
+        CacheWarmingStrategy.AGENT_POOLS: "Pre-initialize agent pool connections",
+        CacheWarmingStrategy.STATIC_RESOURCES: "Cache static configuration and reference data",
+        CacheWarmingStrategy.PREDICTIVE: "Use ML to predict and cache likely needed data"
+    }
+    return descriptions.get(strategy, "No description available")

src/api/routes/admin/database_optimization.py

@@ -0,0 +1,267 @@
+"""
+Module: api.routes.admin.database_optimization
+Description: Admin routes for database optimization
+Author: Anderson H. Silva
+Date: 2025-01-25
+License: Proprietary - All rights reserved
+"""
+
+from typing import List, Optional, Dict, Any
+from fastapi import APIRouter, Depends, HTTPException, status, Query
+
+from src.core import get_logger
+from src.api.dependencies import require_admin, get_db
+from src.services.database_optimization_service import database_optimization_service
+
+logger = get_logger(__name__)
+
+router = APIRouter(prefix="/database-optimization", tags=["Admin - Database Optimization"])
+
+
+@router.get("/analyze-slow-queries")
+async def analyze_slow_queries(
+    limit: int = Query(default=20, ge=1, le=100),
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Analyze slow queries and get optimization suggestions.
+    
+    Requires admin privileges.
+    """
+    try:
+        analyses = await database_optimization_service.analyze_slow_queries(
+            session=db,
+            limit=limit
+        )
+        
+        # Format response
+        results = []
+        for analysis in analyses:
+            results.append({
+                "query": analysis.query[:200] + "..." if len(analysis.query) > 200 else analysis.query,
+                "execution_time": analysis.execution_time,
+                "calls": analysis.plan.get("calls", 0),
+                "total_time": analysis.plan.get("total_time", 0),
+                "suggestions": analysis.suggestions,
+                "estimated_improvement": analysis.estimated_improvement
+            })
+        
+        logger.info(
+            "admin_slow_queries_analyzed",
+            admin=admin_user.get("email"),
+            queries_count=len(results)
+        )
+        
+        return {
+            "slow_queries": results,
+            "total": len(results),
+            "threshold_seconds": database_optimization_service._slow_query_threshold
+        }
+        
+    except Exception as e:
+        logger.error(
+            "analyze_slow_queries_error",
+            error=str(e),
+            exc_info=True
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to analyze slow queries"
+        )
+
+
+@router.get("/missing-indexes")
+async def get_missing_indexes(
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Get suggestions for missing indexes.
+    
+    Requires admin privileges.
+    """
+    try:
+        index_suggestions = await database_optimization_service.create_missing_indexes(
+            session=db,
+            dry_run=True  # Don't create, just suggest
+        )
+        
+        logger.info(
+            "admin_missing_indexes_analyzed",
+            admin=admin_user.get("email"),
+            suggestions_count=len(index_suggestions)
+        )
+        
+        return {
+            "missing_indexes": index_suggestions,
+            "total": len(index_suggestions)
+        }
+        
+    except Exception as e:
+        logger.error(
+            "missing_indexes_error",
+            error=str(e),
+            exc_info=True
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to analyze missing indexes"
+        )
+
+
+@router.post("/create-indexes")
+async def create_missing_indexes(
+    dry_run: bool = Query(default=True, description="If true, only show what would be created"),
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Create missing indexes based on analysis.
+    
+    Requires admin privileges.
+    """
+    try:
+        results = await database_optimization_service.create_missing_indexes(
+            session=db,
+            dry_run=dry_run
+        )
+        
+        created_count = sum(1 for r in results if r.get("status") == "created")
+        
+        logger.info(
+            "admin_indexes_created",
+            admin=admin_user.get("email"),
+            dry_run=dry_run,
+            created=created_count,
+            total=len(results)
+        )
+        
+        return {
+            "dry_run": dry_run,
+            "indexes": results,
+            "created": created_count,
+            "total": len(results)
+        }
+        
+    except Exception as e:
+        logger.error(
+            "create_indexes_error",
+            error=str(e),
+            exc_info=True
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to create indexes"
+        )
+
+
+@router.post("/optimize-statistics")
+async def optimize_table_statistics(
+    tables: Optional[List[str]] = None,
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Update table statistics for query planner optimization.
+    
+    Requires admin privileges.
+    """
+    try:
+        results = await database_optimization_service.optimize_table_statistics(
+            session=db,
+            tables=tables
+        )
+        
+        logger.info(
+            "admin_statistics_optimized",
+            admin=admin_user.get("email"),
+            analyzed=len(results["analyzed"]),
+            vacuumed=len(results["vacuumed"])
+        )
+        
+        return results
+        
+    except Exception as e:
+        logger.error(
+            "optimize_statistics_error",
+            error=str(e),
+            exc_info=True
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to optimize statistics"
+        )
+
+
+@router.get("/database-stats")
+async def get_database_statistics(
+    admin_user=Depends(require_admin),
+    db=Depends(get_db)
+):
+    """
+    Get comprehensive database statistics.
+    
+    Requires admin privileges.
+    """
+    try:
+        stats = await database_optimization_service.get_database_stats(db)
+        
+        return {
+            "database_size": stats.get("database_size"),
+            "largest_tables": stats.get("largest_tables", []),
+            "least_used_indexes": stats.get("least_used_indexes", []),
+            "cache_hit_ratio": stats.get("cache_hit_ratio"),
+            "connections": stats.get("connections"),
+            "recommendations": generate_recommendations(stats)
+        }
+        
+    except Exception as e:
+        logger.error(
+            "database_stats_error",
+            error=str(e),
+            exc_info=True
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to get database statistics"
+        )
+
+
+def generate_recommendations(stats: Dict[str, Any]) -> List[str]:
+    """Generate recommendations based on statistics."""
+    recommendations = []
+    
+    # Cache hit ratio
+    cache_ratio = stats.get("cache_hit_ratio", {}).get("ratio", 0)
+    if cache_ratio < 0.90:
+        recommendations.append(
+            f"Cache hit ratio is {cache_ratio:.1%}. Consider increasing shared_buffers."
+        )
+    
+    # Unused indexes
+    unused_indexes = [
+        idx for idx in stats.get("least_used_indexes", [])
+        if idx["scans"] == 0
+    ]
+    if unused_indexes:
+        recommendations.append(
+            f"Found {len(unused_indexes)} unused indexes consuming space. Consider dropping them."
+        )
+    
+    # Connection pooling
+    connections = stats.get("connections", {})
+    idle_ratio = connections.get("idle", 0) / max(connections.get("total", 1), 1)
+    if idle_ratio > 0.8:
+        recommendations.append(
+            "High idle connection ratio. Consider adjusting connection pool settings."
+        )
+    
+    # Large tables
+    large_tables = stats.get("largest_tables", [])
+    if large_tables and large_tables[0]["row_count"] > 10000000:
+        recommendations.append(
+            "Very large tables detected. Consider partitioning for better performance."
+        )
+    
+    return recommendations

src/services/cache_warming_service.py

@@ -0,0 +1,454 @@
+"""
+Module: services.cache_warming_service
+Description: Cache warming strategies for improved performance
+Author: Anderson H. Silva
+Date: 2025-01-25
+License: Proprietary - All rights reserved
+"""
+
+import asyncio
+from typing import List, Dict, Any, Optional, Set
+from datetime import datetime, timedelta, timezone
+from enum import Enum
+import hashlib
+
+from src.core import get_logger
+from src.infrastructure.cache import cache_service
+from src.services.data_service import data_service
+from src.services.investigation_service import investigation_service
+from src.core.config import settings
+from src.db.session import get_session
+from src.models import Investigation
+
+logger = get_logger(__name__)
+
+
+class CacheWarmingStrategy(str, Enum):
+    """Cache warming strategies."""
+    POPULAR_DATA = "popular_data"
+    RECENT_INVESTIGATIONS = "recent_investigations"
+    FREQUENT_QUERIES = "frequent_queries"
+    AGENT_POOLS = "agent_pools"
+    STATIC_RESOURCES = "static_resources"
+    PREDICTIVE = "predictive"
+
+
+class CacheWarmingConfig:
+    """Configuration for cache warming."""
+    
+    # TTLs por tipo de dado
+    TTL_CONFIG = {
+        "contracts": 3600,          # 1 hora
+        "investigations": 1800,     # 30 minutos
+        "agent_pools": 7200,        # 2 horas
+        "static_data": 86400,       # 24 horas
+        "frequent_queries": 600,    # 10 minutos
+        "analytics": 3600          # 1 hora
+    }
+    
+    # Limites de warming
+    MAX_ITEMS_PER_TYPE = {
+        "contracts": 100,
+        "investigations": 50,
+        "queries": 200,
+        "agents": 20
+    }
+    
+    # Configuração de prioridades
+    PRIORITY_WEIGHTS = {
+        "recency": 0.3,
+        "frequency": 0.4,
+        "importance": 0.3
+    }
+
+
+class CacheWarmingService:
+    """Service for cache warming operations."""
+    
+    def __init__(self):
+        """Initialize cache warming service."""
+        self._config = CacheWarmingConfig()
+        self._warming_tasks: Set[asyncio.Task] = set()
+        self._last_warming: Dict[str, datetime] = {}
+        self._query_frequency: Dict[str, int] = {}
+        self._warming_interval = 300  # 5 minutos
+        
+    async def start_warming_scheduler(self):
+        """Start the cache warming scheduler."""
+        logger.info("cache_warming_scheduler_started")
+        
+        while True:
+            try:
+                # Execute warming strategies
+                await self.warm_all_caches()
+                
+                # Wait for next interval
+                await asyncio.sleep(self._warming_interval)
+                
+            except asyncio.CancelledError:
+                logger.info("cache_warming_scheduler_stopped")
+                break
+            except Exception as e:
+                logger.error(
+                    "cache_warming_scheduler_error",
+                    error=str(e),
+                    exc_info=True
+                )
+                await asyncio.sleep(60)  # Wait 1 minute on error
+    
+    async def warm_all_caches(self):
+        """Execute all cache warming strategies."""
+        start_time = datetime.now(timezone.utc)
+        
+        strategies = [
+            self._warm_popular_data(),
+            self._warm_recent_investigations(),
+            self._warm_frequent_queries(),
+            self._warm_agent_pools(),
+            self._warm_static_resources()
+        ]
+        
+        # Execute strategies in parallel
+        results = await asyncio.gather(*strategies, return_exceptions=True)
+        
+        # Log results
+        duration = (datetime.now(timezone.utc) - start_time).total_seconds()
+        successful = sum(1 for r in results if not isinstance(r, Exception))
+        
+        logger.info(
+            "cache_warming_completed",
+            duration_seconds=duration,
+            strategies_total=len(strategies),
+            strategies_successful=successful
+        )
+        
+        # Update last warming time
+        self._last_warming["all"] = datetime.now(timezone.utc)
+    
+    async def _warm_popular_data(self) -> Dict[str, Any]:
+        """Warm cache with popular/frequently accessed data."""
+        warmed_count = 0
+        
+        try:
+            # Get most accessed contracts
+            popular_contracts = await self._get_popular_contracts()
+            
+            for contract_id in popular_contracts[:self._config.MAX_ITEMS_PER_TYPE["contracts"]]:
+                cache_key = f"contract:{contract_id}"
+                
+                # Check if already cached
+                if await cache_service.get(cache_key):
+                    continue
+                
+                # Fetch and cache
+                try:
+                    contract_data = await data_service.get_contract(contract_id)
+                    if contract_data:
+                        await cache_service.set(
+                            cache_key,
+                            contract_data,
+                            ttl=self._config.TTL_CONFIG["contracts"]
+                        )
+                        warmed_count += 1
+                except Exception as e:
+                    logger.error(f"Failed to warm contract {contract_id}: {e}")
+            
+            logger.info(
+                "popular_data_warmed",
+                contracts_warmed=warmed_count
+            )
+            
+            return {"contracts": warmed_count}
+            
+        except Exception as e:
+            logger.error("popular_data_warming_failed", error=str(e))
+            raise
+    
+    async def _warm_recent_investigations(self) -> Dict[str, Any]:
+        """Warm cache with recent investigations."""
+        warmed_count = 0
+        
+        try:
+            async with get_session() as session:
+                # Get recent investigations
+                from sqlalchemy import select, desc
+                query = select(Investigation).order_by(
+                    desc(Investigation.created_at)
+                ).limit(self._config.MAX_ITEMS_PER_TYPE["investigations"])
+                
+                result = await session.execute(query)
+                investigations = result.scalars().all()
+                
+                for investigation in investigations:
+                    cache_key = f"investigation:{investigation.id}"
+                    
+                    # Cache investigation data
+                    await cache_service.set(
+                        cache_key,
+                        {
+                            "id": investigation.id,
+                            "status": investigation.status,
+                            "contract_id": investigation.contract_id,
+                            "results": investigation.results,
+                            "created_at": investigation.created_at.isoformat()
+                        },
+                        ttl=self._config.TTL_CONFIG["investigations"]
+                    )
+                    warmed_count += 1
+            
+            logger.info(
+                "recent_investigations_warmed",
+                count=warmed_count
+            )
+            
+            return {"investigations": warmed_count}
+            
+        except Exception as e:
+            logger.error("recent_investigations_warming_failed", error=str(e))
+            raise
+    
+    async def _warm_frequent_queries(self) -> Dict[str, Any]:
+        """Warm cache with results of frequent queries."""
+        warmed_count = 0
+        
+        try:
+            # Sort queries by frequency
+            frequent_queries = sorted(
+                self._query_frequency.items(),
+                key=lambda x: x[1],
+                reverse=True
+            )[:self._config.MAX_ITEMS_PER_TYPE["queries"]]
+            
+            for query_hash, frequency in frequent_queries:
+                cache_key = f"query_result:{query_hash}"
+                
+                # Skip if already cached
+                if await cache_service.get(cache_key):
+                    continue
+                
+                # Note: In a real implementation, you would store and replay
+                # the actual query parameters to regenerate results
+                warmed_count += 1
+            
+            logger.info(
+                "frequent_queries_warmed",
+                count=warmed_count
+            )
+            
+            return {"queries": warmed_count}
+            
+        except Exception as e:
+            logger.error("frequent_queries_warming_failed", error=str(e))
+            raise
+    
+    async def _warm_agent_pools(self) -> Dict[str, Any]:
+        """Warm agent pool connections."""
+        warmed_count = 0
+        
+        try:
+            # Pre-initialize agent pools
+            agent_types = [
+                "zumbi",
+                "anita",
+                "tiradentes",
+                "machado",
+                "dandara"
+            ]
+            
+            for agent_type in agent_types[:self._config.MAX_ITEMS_PER_TYPE["agents"]]:
+                cache_key = f"agent_pool:{agent_type}:status"
+                
+                # Cache agent pool status
+                await cache_service.set(
+                    cache_key,
+                    {
+                        "type": agent_type,
+                        "initialized": True,
+                        "last_used": datetime.now(timezone.utc).isoformat()
+                    },
+                    ttl=self._config.TTL_CONFIG["agent_pools"]
+                )
+                warmed_count += 1
+            
+            logger.info(
+                "agent_pools_warmed",
+                count=warmed_count
+            )
+            
+            return {"agents": warmed_count}
+            
+        except Exception as e:
+            logger.error("agent_pools_warming_failed", error=str(e))
+            raise
+    
+    async def _warm_static_resources(self) -> Dict[str, Any]:
+        """Warm cache with static resources."""
+        warmed_count = 0
+        
+        try:
+            # Static data to cache
+            static_data = {
+                "system_config": {
+                    "version": "1.0.0",
+                    "features": ["investigations", "reports", "analysis"],
+                    "agents": ["zumbi", "anita", "tiradentes"]
+                },
+                "contract_types": [
+                    "licitacao",
+                    "contrato",
+                    "convenio",
+                    "termo_aditivo"
+                ],
+                "anomaly_types": [
+                    "valor_atipico",
+                    "padrao_temporal",
+                    "fornecedor_suspeito",
+                    "fragmentacao"
+                ]
+            }
+            
+            for key, data in static_data.items():
+                cache_key = f"static:{key}"
+                await cache_service.set(
+                    cache_key,
+                    data,
+                    ttl=self._config.TTL_CONFIG["static_data"]
+                )
+                warmed_count += 1
+            
+            logger.info(
+                "static_resources_warmed",
+                count=warmed_count
+            )
+            
+            return {"static": warmed_count}
+            
+        except Exception as e:
+            logger.error("static_resources_warming_failed", error=str(e))
+            raise
+    
+    async def _get_popular_contracts(self) -> List[str]:
+        """Get list of popular contract IDs."""
+        # In a real implementation, this would query analytics
+        # or access logs to find most accessed contracts
+        return [
+            "CONT-2024-001",
+            "CONT-2024-002",
+            "CONT-2024-003",
+            "CONT-2024-004",
+            "CONT-2024-005"
+        ]
+    
+    def track_query(self, query_params: Dict[str, Any]):
+        """Track query frequency for cache warming."""
+        # Generate query hash
+        query_str = str(sorted(query_params.items()))
+        query_hash = hashlib.md5(query_str.encode()).hexdigest()
+        
+        # Update frequency
+        self._query_frequency[query_hash] = self._query_frequency.get(query_hash, 0) + 1
+        
+        # Limit stored queries
+        if len(self._query_frequency) > 1000:
+            # Remove least frequent queries
+            sorted_queries = sorted(
+                self._query_frequency.items(),
+                key=lambda x: x[1]
+            )
+            for query, _ in sorted_queries[:100]:
+                del self._query_frequency[query]
+    
+    async def warm_specific_data(
+        self,
+        data_type: str,
+        identifiers: List[str],
+        ttl: Optional[int] = None
+    ) -> Dict[str, Any]:
+        """Warm cache with specific data."""
+        if ttl is None:
+            ttl = self._config.TTL_CONFIG.get(data_type, 3600)
+        
+        warmed = []
+        failed = []
+        
+        for identifier in identifiers:
+            try:
+                cache_key = f"{data_type}:{identifier}"
+                
+                # Skip if already cached
+                if await cache_service.get(cache_key):
+                    continue
+                
+                # Fetch data based on type
+                data = None
+                if data_type == "contract":
+                    data = await data_service.get_contract(identifier)
+                elif data_type == "investigation":
+                    data = await investigation_service.get_investigation(identifier)
+                
+                if data:
+                    await cache_service.set(cache_key, data, ttl=ttl)
+                    warmed.append(identifier)
+                else:
+                    failed.append(identifier)
+                    
+            except Exception as e:
+                logger.error(
+                    f"Failed to warm {data_type}:{identifier}: {e}"
+                )
+                failed.append(identifier)
+        
+        return {
+            "warmed": warmed,
+            "failed": failed,
+            "total": len(identifiers)
+        }
+    
+    async def get_warming_status(self) -> Dict[str, Any]:
+        """Get current cache warming status."""
+        status = {
+            "last_warming": self._last_warming.get("all"),
+            "query_frequency_tracked": len(self._query_frequency),
+            "top_queries": sorted(
+                self._query_frequency.items(),
+                key=lambda x: x[1],
+                reverse=True
+            )[:10],
+            "config": {
+                "interval_seconds": self._warming_interval,
+                "ttls": self._config.TTL_CONFIG,
+                "limits": self._config.MAX_ITEMS_PER_TYPE
+            }
+        }
+        
+        return status
+    
+    async def trigger_manual_warming(
+        self,
+        strategies: Optional[List[CacheWarmingStrategy]] = None
+    ) -> Dict[str, Any]:
+        """Manually trigger cache warming."""
+        if strategies is None:
+            return await self.warm_all_caches()
+        
+        results = {}
+        for strategy in strategies:
+            try:
+                if strategy == CacheWarmingStrategy.POPULAR_DATA:
+                    results[strategy] = await self._warm_popular_data()
+                elif strategy == CacheWarmingStrategy.RECENT_INVESTIGATIONS:
+                    results[strategy] = await self._warm_recent_investigations()
+                elif strategy == CacheWarmingStrategy.FREQUENT_QUERIES:
+                    results[strategy] = await self._warm_frequent_queries()
+                elif strategy == CacheWarmingStrategy.AGENT_POOLS:
+                    results[strategy] = await self._warm_agent_pools()
+                elif strategy == CacheWarmingStrategy.STATIC_RESOURCES:
+                    results[strategy] = await self._warm_static_resources()
+            except Exception as e:
+                results[strategy] = {"error": str(e)}
+        
+        return results
+
+
+# Global instance
+cache_warming_service = CacheWarmingService()

src/services/database_optimization_service.py

@@ -0,0 +1,576 @@
+"""
+Module: services.database_optimization_service
+Description: Database query optimization and index management
+Author: Anderson H. Silva
+Date: 2025-01-25
+License: Proprietary - All rights reserved
+"""
+
+from typing import List, Dict, Any, Optional, Tuple
+from datetime import datetime, timedelta, timezone
+from sqlalchemy import text, create_engine, inspect
+from sqlalchemy.ext.asyncio import AsyncSession
+import asyncio
+import time
+
+from src.core import get_logger
+from src.db.session import get_session
+from src.core.config import settings
+
+logger = get_logger(__name__)
+
+
+class QueryAnalysis:
+    """Analysis result for a database query."""
+    
+    def __init__(self, query: str, execution_time: float, plan: Dict[str, Any]):
+        self.query = query
+        self.execution_time = execution_time
+        self.plan = plan
+        self.suggestions = []
+        self.estimated_improvement = 0.0
+    
+    def add_suggestion(self, suggestion: str, improvement: float = 0.0):
+        """Add optimization suggestion."""
+        self.suggestions.append(suggestion)
+        self.estimated_improvement += improvement
+
+
+class DatabaseOptimizationService:
+    """Service for database performance optimization."""
+    
+    def __init__(self):
+        """Initialize database optimization service."""
+        self._slow_query_threshold = 1.0  # seconds
+        self._index_suggestions = {}
+        self._query_stats = {}
+        
+    async def analyze_slow_queries(
+        self,
+        session: AsyncSession,
+        limit: int = 20
+    ) -> List[QueryAnalysis]:
+        """Analyze slow queries from PostgreSQL."""
+        analyses = []
+        
+        try:
+            # Get slow queries from pg_stat_statements
+            slow_queries_sql = """
+            SELECT 
+                query,
+                mean_exec_time / 1000.0 as mean_exec_seconds,
+                calls,
+                total_exec_time / 1000.0 as total_exec_seconds,
+                min_exec_time / 1000.0 as min_exec_seconds,
+                max_exec_time / 1000.0 as max_exec_seconds,
+                rows
+            FROM pg_stat_statements
+            WHERE mean_exec_time > :threshold_ms
+                AND query NOT LIKE '%pg_stat%'
+                AND query NOT LIKE '%information_schema%'
+            ORDER BY mean_exec_time DESC
+            LIMIT :limit
+            """
+            
+            result = await session.execute(
+                text(slow_queries_sql),
+                {
+                    "threshold_ms": self._slow_query_threshold * 1000,
+                    "limit": limit
+                }
+            )
+            
+            rows = result.fetchall()
+            
+            for row in rows:
+                # Analyze each slow query
+                analysis = QueryAnalysis(
+                    query=row.query,
+                    execution_time=row.mean_exec_seconds,
+                    plan={
+                        "calls": row.calls,
+                        "total_time": row.total_exec_seconds,
+                        "min_time": row.min_exec_seconds,
+                        "max_time": row.max_exec_seconds,
+                        "rows": row.rows
+                    }
+                )
+                
+                # Get query plan
+                await self._analyze_query_plan(session, analysis)
+                
+                # Generate suggestions
+                self._generate_suggestions(analysis)
+                
+                analyses.append(analysis)
+            
+            logger.info(
+                "slow_query_analysis_completed",
+                queries_analyzed=len(analyses)
+            )
+            
+        except Exception as e:
+            logger.error(
+                "slow_query_analysis_error",
+                error=str(e),
+                exc_info=True
+            )
+        
+        return analyses
+    
+    async def _analyze_query_plan(
+        self,
+        session: AsyncSession,
+        analysis: QueryAnalysis
+    ):
+        """Analyze query execution plan."""
+        try:
+            # Get EXPLAIN ANALYZE for the query
+            explain_sql = f"EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) {analysis.query}"
+            
+            result = await session.execute(text(explain_sql))
+            plan_data = result.scalar()
+            
+            if plan_data:
+                analysis.plan["execution_plan"] = plan_data[0]["Plan"]
+                
+                # Extract key metrics
+                plan = plan_data[0]["Plan"]
+                analysis.plan["total_cost"] = plan.get("Total Cost", 0)
+                analysis.plan["actual_time"] = plan.get("Actual Total Time", 0)
+                
+                # Look for problematic patterns
+                self._check_plan_issues(plan, analysis)
+                
+        except Exception as e:
+            logger.debug(f"Could not analyze plan for query: {e}")
+    
+    def _check_plan_issues(self, plan: Dict[str, Any], analysis: QueryAnalysis):
+        """Check for common plan issues."""
+        # Sequential scan on large tables
+        if plan.get("Node Type") == "Seq Scan":
+            rows = plan.get("Actual Rows", 0)
+            if rows > 1000:
+                analysis.add_suggestion(
+                    f"Sequential scan on {rows} rows. Consider adding an index.",
+                    improvement=0.5
+                )
+        
+        # Nested loops with high iterations
+        if plan.get("Node Type") == "Nested Loop":
+            loops = plan.get("Actual Loops", 0)
+            if loops > 100:
+                analysis.add_suggestion(
+                    f"Nested loop with {loops} iterations. Consider query restructuring.",
+                    improvement=0.3
+                )
+        
+        # Check child nodes recursively
+        if "Plans" in plan:
+            for child_plan in plan["Plans"]:
+                self._check_plan_issues(child_plan, analysis)
+    
+    def _generate_suggestions(self, analysis: QueryAnalysis):
+        """Generate optimization suggestions for a query."""
+        query_lower = analysis.query.lower()
+        
+        # Check for missing LIMIT
+        if "select" in query_lower and "limit" not in query_lower:
+            if analysis.plan.get("rows", 0) > 1000:
+                analysis.add_suggestion(
+                    "Query returns many rows. Consider adding LIMIT clause.",
+                    improvement=0.2
+                )
+        
+        # Check for SELECT *
+        if "select *" in query_lower:
+            analysis.add_suggestion(
+                "Avoid SELECT *. Specify only needed columns.",
+                improvement=0.1
+            )
+        
+        # Check for missing WHERE on large tables
+        if "where" not in query_lower and analysis.plan.get("rows", 0) > 10000:
+            analysis.add_suggestion(
+                "No WHERE clause on large result set. Add filtering.",
+                improvement=0.4
+            )
+        
+        # Check for IN with many values
+        import re
+        in_matches = re.findall(r'IN\s*\([^)]+\)', query_lower)
+        for match in in_matches:
+            values_count = match.count(',') + 1
+            if values_count > 10:
+                analysis.add_suggestion(
+                    f"IN clause with {values_count} values. Consider using JOIN or temp table.",
+                    improvement=0.2
+                )
+    
+    async def create_missing_indexes(
+        self,
+        session: AsyncSession,
+        dry_run: bool = True
+    ) -> List[Dict[str, Any]]:
+        """Create missing indexes based on analysis."""
+        index_commands = []
+        
+        try:
+            # Analyze foreign key columns without indexes
+            fk_index_sql = """
+            SELECT
+                tc.table_name,
+                kcu.column_name,
+                ccu.table_name AS foreign_table_name
+            FROM information_schema.table_constraints AS tc
+            JOIN information_schema.key_column_usage AS kcu
+                ON tc.constraint_name = kcu.constraint_name
+            JOIN information_schema.constraint_column_usage AS ccu
+                ON ccu.constraint_name = tc.constraint_name
+            WHERE tc.constraint_type = 'FOREIGN KEY'
+                AND NOT EXISTS (
+                    SELECT 1
+                    FROM pg_indexes
+                    WHERE schemaname = 'public'
+                        AND tablename = tc.table_name
+                        AND indexdef LIKE '%' || kcu.column_name || '%'
+                )
+            """
+            
+            result = await session.execute(text(fk_index_sql))
+            fk_without_index = result.fetchall()
+            
+            for row in fk_without_index:
+                index_name = f"idx_{row.table_name}_{row.column_name}"
+                index_cmd = f"CREATE INDEX {index_name} ON {row.table_name} ({row.column_name})"
+                
+                index_commands.append({
+                    "type": "foreign_key",
+                    "table": row.table_name,
+                    "column": row.column_name,
+                    "command": index_cmd,
+                    "reason": f"Foreign key to {row.foreign_table_name}"
+                })
+            
+            # Analyze frequently filtered columns
+            filter_columns = await self._analyze_filter_columns(session)
+            
+            for table, column, frequency in filter_columns:
+                # Check if index already exists
+                check_sql = """
+                SELECT 1 FROM pg_indexes
+                WHERE schemaname = 'public'
+                    AND tablename = :table
+                    AND indexdef LIKE :pattern
+                """
+                
+                exists = await session.execute(
+                    text(check_sql),
+                    {"table": table, "pattern": f"%{column}%"}
+                )
+                
+                if not exists.scalar():
+                    index_name = f"idx_{table}_{column}_filter"
+                    index_cmd = f"CREATE INDEX {index_name} ON {table} ({column})"
+                    
+                    index_commands.append({
+                        "type": "frequent_filter",
+                        "table": table,
+                        "column": column,
+                        "command": index_cmd,
+                        "reason": f"Frequently used in WHERE clause ({frequency} times)"
+                    })
+            
+            # Execute or return commands
+            if not dry_run and index_commands:
+                for idx_info in index_commands:
+                    try:
+                        await session.execute(text(idx_info["command"]))
+                        idx_info["status"] = "created"
+                        logger.info(
+                            "index_created",
+                            table=idx_info["table"],
+                            column=idx_info["column"]
+                        )
+                    except Exception as e:
+                        idx_info["status"] = "failed"
+                        idx_info["error"] = str(e)
+                        logger.error(
+                            "index_creation_failed",
+                            table=idx_info["table"],
+                            error=str(e)
+                        )
+                
+                await session.commit()
+            
+        except Exception as e:
+            logger.error(
+                "create_indexes_error",
+                error=str(e),
+                exc_info=True
+            )
+        
+        return index_commands
+    
+    async def _analyze_filter_columns(
+        self,
+        session: AsyncSession
+    ) -> List[Tuple[str, str, int]]:
+        """Analyze frequently filtered columns from query patterns."""
+        filter_columns = []
+        
+        try:
+            # Parse WHERE clauses from pg_stat_statements
+            filter_analysis_sql = """
+            SELECT 
+                query,
+                calls
+            FROM pg_stat_statements
+            WHERE query LIKE '%WHERE%'
+                AND query NOT LIKE '%pg_stat%'
+                AND calls > 10
+            ORDER BY calls DESC
+            LIMIT 100
+            """
+            
+            result = await session.execute(text(filter_analysis_sql))
+            queries = result.fetchall()
+            
+            # Simple pattern matching for WHERE conditions
+            import re
+            column_frequency = {}
+            
+            for query, calls in queries:
+                # Extract table.column or column patterns after WHERE
+                where_match = re.search(r'WHERE\s+(.+?)(?:ORDER|GROUP|LIMIT|$)', query, re.IGNORECASE)
+                if where_match:
+                    conditions = where_match.group(1)
+                    
+                    # Find column references
+                    column_patterns = re.findall(r'(\w+)\.(\w+)\s*[=<>]|(\w+)\s*[=<>]', conditions)
+                    
+                    for pattern in column_patterns:
+                        if pattern[0] and pattern[1]:  # table.column format
+                            key = (pattern[0], pattern[1])
+                        elif pattern[2]:  # column only format
+                            # Try to infer table from FROM clause
+                            from_match = re.search(r'FROM\s+(\w+)', query, re.IGNORECASE)
+                            if from_match:
+                                key = (from_match.group(1), pattern[2])
+                            else:
+                                continue
+                        else:
+                            continue
+                        
+                        column_frequency[key] = column_frequency.get(key, 0) + calls
+            
+            # Sort by frequency
+            for (table, column), frequency in sorted(
+                column_frequency.items(),
+                key=lambda x: x[1],
+                reverse=True
+            )[:20]:
+                filter_columns.append((table, column, frequency))
+            
+        except Exception as e:
+            logger.error(
+                "filter_column_analysis_error",
+                error=str(e),
+                exc_info=True
+            )
+        
+        return filter_columns
+    
+    async def optimize_table_statistics(
+        self,
+        session: AsyncSession,
+        tables: Optional[List[str]] = None
+    ) -> Dict[str, Any]:
+        """Update table statistics for query planner."""
+        results = {
+            "analyzed": [],
+            "vacuumed": [],
+            "errors": []
+        }
+        
+        try:
+            # Get all tables if not specified
+            if not tables:
+                tables_sql = """
+                SELECT tablename 
+                FROM pg_tables 
+                WHERE schemaname = 'public'
+                """
+                result = await session.execute(text(tables_sql))
+                tables = [row[0] for row in result.fetchall()]
+            
+            for table in tables:
+                try:
+                    # ANALYZE table
+                    await session.execute(text(f"ANALYZE {table}"))
+                    results["analyzed"].append(table)
+                    
+                    # Check if VACUUM needed
+                    vacuum_check_sql = """
+                    SELECT 
+                        n_dead_tup,
+                        n_live_tup
+                    FROM pg_stat_user_tables
+                    WHERE relname = :table
+                    """
+                    
+                    result = await session.execute(
+                        text(vacuum_check_sql),
+                        {"table": table}
+                    )
+                    row = result.fetchone()
+                    
+                    if row and row.n_dead_tup > row.n_live_tup * 0.2:
+                        # More than 20% dead tuples, vacuum needed
+                        await session.execute(text(f"VACUUM ANALYZE {table}"))
+                        results["vacuumed"].append(table)
+                        logger.info(
+                            "table_vacuumed",
+                            table=table,
+                            dead_tuples=row.n_dead_tup
+                        )
+                    
+                except Exception as e:
+                    results["errors"].append({
+                        "table": table,
+                        "error": str(e)
+                    })
+                    logger.error(
+                        f"Failed to optimize table {table}: {e}"
+                    )
+            
+            await session.commit()
+            
+        except Exception as e:
+            logger.error(
+                "table_optimization_error",
+                error=str(e),
+                exc_info=True
+            )
+        
+        return results
+    
+    async def get_database_stats(
+        self,
+        session: AsyncSession
+    ) -> Dict[str, Any]:
+        """Get comprehensive database statistics."""
+        stats = {}
+        
+        try:
+            # Database size
+            size_sql = """
+            SELECT 
+                pg_database_size(current_database()) as db_size,
+                pg_size_pretty(pg_database_size(current_database())) as db_size_pretty
+            """
+            result = await session.execute(text(size_sql))
+            size_info = result.fetchone()
+            stats["database_size"] = {
+                "bytes": size_info.db_size,
+                "pretty": size_info.db_size_pretty
+            }
+            
+            # Table sizes
+            table_sizes_sql = """
+            SELECT 
+                schemaname,
+                tablename,
+                pg_total_relation_size(schemaname||'.'||tablename) as total_size,
+                pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as size_pretty,
+                n_live_tup as row_count
+            FROM pg_tables
+            JOIN pg_stat_user_tables USING (schemaname, tablename)
+            WHERE schemaname = 'public'
+            ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC
+            LIMIT 10
+            """
+            result = await session.execute(text(table_sizes_sql))
+            stats["largest_tables"] = [
+                {
+                    "table": row.tablename,
+                    "size_bytes": row.total_size,
+                    "size_pretty": row.size_pretty,
+                    "row_count": row.row_count
+                }
+                for row in result.fetchall()
+            ]
+            
+            # Index usage
+            index_usage_sql = """
+            SELECT 
+                schemaname,
+                tablename,
+                indexname,
+                idx_scan,
+                idx_tup_read,
+                idx_tup_fetch,
+                pg_size_pretty(pg_relation_size(indexrelid)) as index_size
+            FROM pg_stat_user_indexes
+            WHERE schemaname = 'public'
+            ORDER BY idx_scan
+            LIMIT 20
+            """
+            result = await session.execute(text(index_usage_sql))
+            stats["least_used_indexes"] = [
+                {
+                    "table": row.tablename,
+                    "index": row.indexname,
+                    "scans": row.idx_scan,
+                    "size": row.index_size
+                }
+                for row in result.fetchall()
+            ]
+            
+            # Cache hit ratio
+            cache_sql = """
+            SELECT 
+                sum(heap_blks_read) as heap_read,
+                sum(heap_blks_hit) as heap_hit,
+                sum(heap_blks_hit) / NULLIF(sum(heap_blks_hit) + sum(heap_blks_read), 0) as cache_hit_ratio
+            FROM pg_statio_user_tables
+            """
+            result = await session.execute(text(cache_sql))
+            cache_info = result.fetchone()
+            stats["cache_hit_ratio"] = {
+                "ratio": float(cache_info.cache_hit_ratio or 0),
+                "heap_read": cache_info.heap_read,
+                "heap_hit": cache_info.heap_hit
+            }
+            
+            # Connection stats
+            conn_sql = """
+            SELECT 
+                count(*) as total_connections,
+                count(*) FILTER (WHERE state = 'active') as active_connections,
+                count(*) FILTER (WHERE state = 'idle') as idle_connections,
+                count(*) FILTER (WHERE state = 'idle in transaction') as idle_in_transaction
+            FROM pg_stat_activity
+            WHERE datname = current_database()
+            """
+            result = await session.execute(text(conn_sql))
+            conn_info = result.fetchone()
+            stats["connections"] = {
+                "total": conn_info.total_connections,
+                "active": conn_info.active_connections,
+                "idle": conn_info.idle_connections,
+                "idle_in_transaction": conn_info.idle_in_transaction
+            }
+            
+        except Exception as e:
+            logger.error(
+                "database_stats_error",
+                error=str(e),
+                exc_info=True
+            )
+        
+        return stats
+
+
+# Global instance
+database_optimization_service = DatabaseOptimizationService()

src/tasks/cache_warming_tasks.py

@@ -0,0 +1,328 @@
+"""
+Module: tasks.cache_warming_tasks
+Description: Celery tasks for cache warming
+Author: Anderson H. Silva
+Date: 2025-01-25
+License: Proprietary - All rights reserved
+"""
+
+from datetime import datetime, timezone
+from typing import Dict, Any, List, Optional
+
+from celery import shared_task
+from celery.utils.log import get_task_logger
+
+from src.services.cache_warming_service import (
+    cache_warming_service,
+    CacheWarmingStrategy
+)
+
+logger = get_task_logger(__name__)
+
+
+@shared_task(
+    name="cache_warming.warm_all",
+    max_retries=3,
+    default_retry_delay=300,
+    time_limit=600,
+    soft_time_limit=540
+)
+def warm_all_caches() -> Dict[str, Any]:
+    """
+    Warm all caches using all strategies.
+    
+    This task is scheduled to run periodically.
+    """
+    try:
+        logger.info("Starting scheduled cache warming")
+        
+        # Execute warming synchronously
+        import asyncio
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        
+        try:
+            result = loop.run_until_complete(
+                cache_warming_service.warm_all_caches()
+            )
+            
+            logger.info(
+                "Cache warming completed",
+                result=result
+            )
+            
+            return {
+                "status": "completed",
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "result": result
+            }
+            
+        finally:
+            loop.close()
+            
+    except Exception as e:
+        logger.error(
+            f"Cache warming failed: {str(e)}",
+            exc_info=True
+        )
+        raise
+
+
+@shared_task(
+    name="cache_warming.warm_strategy",
+    max_retries=3,
+    default_retry_delay=60
+)
+def warm_specific_strategy(strategy: str) -> Dict[str, Any]:
+    """
+    Warm cache using a specific strategy.
+    
+    Args:
+        strategy: Name of the warming strategy
+    """
+    try:
+        logger.info(f"Starting cache warming for strategy: {strategy}")
+        
+        # Convert string to enum
+        strategy_enum = CacheWarmingStrategy(strategy)
+        
+        # Execute warming
+        import asyncio
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        
+        try:
+            result = loop.run_until_complete(
+                cache_warming_service.trigger_manual_warming([strategy_enum])
+            )
+            
+            return {
+                "status": "completed",
+                "strategy": strategy,
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "result": result
+            }
+            
+        finally:
+            loop.close()
+            
+    except Exception as e:
+        logger.error(
+            f"Strategy warming failed: {str(e)}",
+            exc_info=True
+        )
+        raise
+
+
+@shared_task(
+    name="cache_warming.warm_contracts",
+    max_retries=2,
+    default_retry_delay=120
+)
+def warm_contract_cache(contract_ids: List[str]) -> Dict[str, Any]:
+    """
+    Warm cache for specific contracts.
+    
+    Args:
+        contract_ids: List of contract IDs to cache
+    """
+    try:
+        logger.info(
+            f"Warming cache for {len(contract_ids)} contracts"
+        )
+        
+        import asyncio
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        
+        try:
+            result = loop.run_until_complete(
+                cache_warming_service.warm_specific_data(
+                    data_type="contract",
+                    identifiers=contract_ids,
+                    ttl=3600  # 1 hour
+                )
+            )
+            
+            logger.info(
+                f"Contract cache warming completed: "
+                f"{len(result['warmed'])} warmed, "
+                f"{len(result['failed'])} failed"
+            )
+            
+            return result
+            
+        finally:
+            loop.close()
+            
+    except Exception as e:
+        logger.error(
+            f"Contract cache warming failed: {str(e)}",
+            exc_info=True
+        )
+        raise
+
+
+@shared_task(
+    name="cache_warming.warm_investigations",
+    max_retries=2,
+    default_retry_delay=120
+)
+def warm_investigation_cache(
+    investigation_ids: Optional[List[str]] = None,
+    limit: int = 50
+) -> Dict[str, Any]:
+    """
+    Warm cache for investigations.
+    
+    Args:
+        investigation_ids: Specific IDs or None for recent
+        limit: Maximum number to warm if no IDs provided
+    """
+    try:
+        import asyncio
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        
+        try:
+            if investigation_ids:
+                # Warm specific investigations
+                result = loop.run_until_complete(
+                    cache_warming_service.warm_specific_data(
+                        data_type="investigation",
+                        identifiers=investigation_ids,
+                        ttl=1800  # 30 minutes
+                    )
+                )
+            else:
+                # Warm recent investigations
+                result = loop.run_until_complete(
+                    cache_warming_service.trigger_manual_warming(
+                        [CacheWarmingStrategy.RECENT_INVESTIGATIONS]
+                    )
+                )
+            
+            return {
+                "status": "completed",
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "result": result
+            }
+            
+        finally:
+            loop.close()
+            
+    except Exception as e:
+        logger.error(
+            f"Investigation cache warming failed: {str(e)}",
+            exc_info=True
+        )
+        raise
+
+
+@shared_task(
+    name="cache_warming.analyze_patterns",
+    max_retries=1,
+    time_limit=300
+)
+def analyze_cache_patterns() -> Dict[str, Any]:
+    """
+    Analyze cache access patterns for optimization.
+    
+    This task collects metrics about cache usage to improve
+    warming strategies.
+    """
+    try:
+        logger.info("Analyzing cache access patterns")
+        
+        import asyncio
+        from src.infrastructure.cache import cache_service
+        
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        
+        try:
+            # Get cache statistics
+            stats = loop.run_until_complete(
+                cache_service.get_stats()
+            )
+            
+            # Get warming status
+            warming_status = loop.run_until_complete(
+                cache_warming_service.get_warming_status()
+            )
+            
+            # Analyze patterns
+            analysis = {
+                "cache_stats": stats,
+                "warming_status": warming_status,
+                "recommendations": []
+            }
+            
+            # Generate recommendations
+            if stats.get("hit_rate", 0) < 0.7:
+                analysis["recommendations"].append(
+                    "Low hit rate detected. Consider warming more frequently accessed data."
+                )
+            
+            if warming_status["query_frequency_tracked"] > 500:
+                analysis["recommendations"].append(
+                    "High query diversity. Consider implementing predictive warming."
+                )
+            
+            logger.info(
+                "Cache pattern analysis completed",
+                recommendations=len(analysis["recommendations"])
+            )
+            
+            return analysis
+            
+        finally:
+            loop.close()
+            
+    except Exception as e:
+        logger.error(
+            f"Cache pattern analysis failed: {str(e)}",
+            exc_info=True
+        )
+        raise
+
+
+# Celery Beat schedule configuration
+from celery.schedules import crontab
+
+beat_schedule = {
+    'warm-all-caches': {
+        'task': 'cache_warming.warm_all',
+        'schedule': crontab(minute='*/10'),  # Every 10 minutes
+        'options': {
+            'queue': 'cache',
+            'priority': 3
+        }
+    },
+    'warm-popular-data': {
+        'task': 'cache_warming.warm_strategy',
+        'schedule': crontab(minute='*/5'),  # Every 5 minutes
+        'args': ['popular_data'],
+        'options': {
+            'queue': 'cache',
+            'priority': 5
+        }
+    },
+    'warm-static-resources': {
+        'task': 'cache_warming.warm_strategy',
+        'schedule': crontab(hour='*/6'),  # Every 6 hours
+        'args': ['static_resources'],
+        'options': {
+            'queue': 'cache',
+            'priority': 2
+        }
+    },
+    'analyze-cache-patterns': {
+        'task': 'cache_warming.analyze_patterns',
+        'schedule': crontab(hour=2, minute=0),  # Daily at 2 AM
+        'options': {
+            'queue': 'analytics',
+            'priority': 1
+        }
+    }
+}