core/process_manager.py

"""
============================================
RUHI-CORE - Process Manager (THE HEART)
Auto-restart, Health Check, Live Logs
============================================
"""

import os
import sys
import json
import uuid
import signal
import asyncio
import subprocess
import time
from datetime import datetime
from typing import Dict, Optional, List
from pathlib import Path
from collections import deque

import psutil
import aiosqlite
from loguru import logger

from core.config import settings
from core.port_manager import port_manager


class ServiceProcess:
    """Represents a single managed service/process"""
    
    def __init__(self, service_id: str, name: str, **kwargs):
        self.id = service_id
        self.name = name
        self.type = kwargs.get("type", "web")  # web, worker, bot, cron
        self.language = kwargs.get("language", "python")
        self.entry_file = kwargs.get("entry_file", "main.py")
        self.command = kwargs.get("command", "")
        self.app_dir = kwargs.get("app_dir", str(settings.APPS_DIR / name))
        self.port = kwargs.get("port", None)
        self.env_vars = kwargs.get("env_vars", {})
        self.auto_restart = kwargs.get("auto_restart", True)
        self.max_memory_mb = kwargs.get("max_memory_mb", 512)
        self.description = kwargs.get("description", "")
        
        # Runtime state
        self.process: Optional[subprocess.Popen] = None
        self.pid: int = 0
        self.status: str = "stopped"  # stopped, starting, running, crashed, restarting
        self.restart_count: int = 0
        self.last_started: Optional[datetime] = None
        self.last_stopped: Optional[datetime] = None
        self.uptime_start: Optional[float] = None
        
        # Log buffer (keep last 1000 lines in memory)
        self.log_buffer: deque = deque(maxlen=1000)
        self._log_task: Optional[asyncio.Task] = None
        self._monitor_task: Optional[asyncio.Task] = None
    
    def _build_command(self) -> List[str]:
        """Build the command to run the service"""
        if self.command:
            return self.command.split()
        
        entry_path = os.path.join(self.app_dir, self.entry_file)
        
        if self.language == "python":
            return [sys.executable, "-u", entry_path]
        elif self.language == "node" or self.language == "javascript":
            return ["node", entry_path]
        elif self.language == "shell" or self.language == "bash":
            return ["bash", entry_path]
        else:
            return [sys.executable, "-u", entry_path]
    
    def _build_env(self) -> dict:
        """Build environment variables for the process"""
        env = os.environ.copy()
        env.update({
            "RUHI_SERVICE_ID": self.id,
            "RUHI_SERVICE_NAME": self.name,
            "RUHI_SERVICE_TYPE": self.type,
            "RUHI_DATA_DIR": str(settings.DATA_DIR),
            "PORT": str(self.port) if self.port else "8000",
        })
        # Add custom env vars
        if isinstance(self.env_vars, dict):
            env.update(self.env_vars)
        elif isinstance(self.env_vars, str):
            try:
                env.update(json.loads(self.env_vars))
            except json.JSONDecodeError:
                pass
        return env
    
    async def start(self) -> bool:
        """Start the service process"""
        if self.status == "running" and self.process and self.process.poll() is None:
            logger.warning(f"⚠️ Service '{self.name}' is already running (PID: {self.pid})")
            return True
        
        # Allocate port for web services
        if self.type == "web" and not self.port:
            self.port = port_manager.allocate(self.id)
        
        cmd = self._build_command()
        env = self._build_env()
        
        # Ensure app directory exists
        os.makedirs(self.app_dir, exist_ok=True)
        
        # Check if entry file exists
        if not self.command:
            entry_path = os.path.join(self.app_dir, self.entry_file)
            if not os.path.exists(entry_path):
                error_msg = f"Entry file not found: {entry_path}"
                logger.error(f"❌ {error_msg}")
                self.log_buffer.append(f"[ERROR] {error_msg}")
                self.status = "crashed"
                return False
        
        try:
            self.status = "starting"
            log_msg = f"🚀 Starting service '{self.name}' with command: {' '.join(cmd)}"
            logger.info(log_msg)
            self.log_buffer.append(f"[SYSTEM] {log_msg}")
            
            # Create log file
            log_file_path = settings.LOGS_DIR / f"{self.name}.log"
            
            self.process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                cwd=self.app_dir,
                env=env,
                bufsize=1,
                universal_newlines=True,
                preexec_fn=os.setsid if os.name != 'nt' else None
            )
            
            self.pid = self.process.pid
            self.status = "running"
            self.last_started = datetime.now()
            self.uptime_start = time.time()
            
            logger.info(f"✅ Service '{self.name}' started successfully (PID: {self.pid}, Port: {self.port})")
            self.log_buffer.append(f"[SYSTEM] ✅ Started successfully (PID: {self.pid})")
            
            # Start log reader task
            self._log_task = asyncio.create_task(self._read_logs())
            
            # Start monitor task
            self._monitor_task = asyncio.create_task(self._monitor_process())
            
            # Save to DB
            await self._save_to_db()
            
            return True
            
        except Exception as e:
            error_msg = f"Failed to start service '{self.name}': {str(e)}"
            logger.error(f"❌ {error_msg}")
            self.log_buffer.append(f"[ERROR] {error_msg}")
            self.status = "crashed"
            return False
    
    async def stop(self) -> bool:
        """Stop the service process"""
        if not self.process:
            self.status = "stopped"
            return True
        
        try:
            logger.info(f"🛑 Stopping service '{self.name}' (PID: {self.pid})...")
            self.log_buffer.append(f"[SYSTEM] 🛑 Stopping service...")
            
            # Cancel tasks
            if self._log_task and not self._log_task.done():
                self._log_task.cancel()
            if self._monitor_task and not self._monitor_task.done():
                self._monitor_task.cancel()
            
            # Graceful shutdown
            try:
                if os.name != 'nt':
                    os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
                else:
                    self.process.terminate()
            except ProcessLookupError:
                pass
            
            # Wait for process to end (max 10 seconds)
            try:
                self.process.wait(timeout=10)
            except subprocess.TimeoutExpired:
                logger.warning(f"⚠️ Force killing service '{self.name}'")
                try:
                    if os.name != 'nt':
                        os.killpg(os.getpgid(self.process.pid), signal.SIGKILL)
                    else:
                        self.process.kill()
                except ProcessLookupError:
                    pass
            
            self.status = "stopped"
            self.last_stopped = datetime.now()
            self.process = None
            self.pid = 0
            
            # Release port
            if self.type == "web":
                port_manager.release(self.id)
            
            logger.info(f"✅ Service '{self.name}' stopped successfully")
            self.log_buffer.append(f"[SYSTEM] ✅ Service stopped")
            
            await self._save_to_db()
            return True
            
        except Exception as e:
            logger.error(f"❌ Error stopping service '{self.name}': {str(e)}")
            self.status = "crashed"
            return False
    
    async def restart(self) -> bool:
        """Restart the service"""
        self.status = "restarting"
        self.log_buffer.append(f"[SYSTEM] 🔄 Restarting service...")
        await self.stop()
        await asyncio.sleep(2)
        self.restart_count += 1
        return await self.start()
    
    async def _read_logs(self):
        """Continuously read stdout/stderr from the process"""
        try:
            log_file = settings.LOGS_DIR / f"{self.name}.log"
            while self.process and self.process.poll() is None:
                line = await asyncio.get_event_loop().run_in_executor(
                    None, self.process.stdout.readline
                )
                if line:
                    line = line.strip()
                    timestamp = datetime.now().strftime("%H:%M:%S")
                    formatted = f"[{timestamp}] {line}"
                    self.log_buffer.append(formatted)
                    
                    # Write to log file
                    try:
                        with open(log_file, "a", encoding="utf-8") as f:
                            f.write(formatted + "\n")
                    except Exception:
                        pass
                else:
                    await asyncio.sleep(0.1)
        except asyncio.CancelledError:
            pass
        except Exception as e:
            logger.error(f"Log reader error for '{self.name}': {str(e)}")
    
    async def _monitor_process(self):
        """Monitor process health and auto-restart if needed"""
        try:
            while True:
                await asyncio.sleep(settings.HEALTH_CHECK_INTERVAL)
                
                if self.process is None:
                    break
                
                # Check if process is still running
                poll_result = self.process.poll()
                if poll_result is not None:
                    exit_code = poll_result
                    logger.warning(f"💀 Service '{self.name}' exited with code {exit_code}")
                    self.log_buffer.append(f"[SYSTEM] 💀 Process exited with code {exit_code}")
                    self.status = "crashed"
                    
                    # Auto-restart logic
                    if self.auto_restart and self.restart_count < settings.MAX_RESTART_ATTEMPTS:
                        logger.info(f"🔄 Auto-restarting '{self.name}' in {settings.AUTO_RESTART_DELAY}s (attempt {self.restart_count + 1})")
                        self.log_buffer.append(f"[SYSTEM] 🔄 Auto-restarting in {settings.AUTO_RESTART_DELAY}s...")
                        await asyncio.sleep(settings.AUTO_RESTART_DELAY)
                        self.restart_count += 1
                        self.process = None
                        await self.start()
                    else:
                        if self.restart_count >= settings.MAX_RESTART_ATTEMPTS:
                            self.log_buffer.append(f"[SYSTEM] ❌ Max restart attempts reached. Service stopped.")
                            logger.error(f"❌ Service '{self.name}' exceeded max restart attempts")
                    break
                
                # Memory check
                try:
                    proc = psutil.Process(self.pid)
                    memory_mb = proc.memory_info().rss / (1024 * 1024)
                    if memory_mb > self.max_memory_mb:
                        logger.warning(f"⚠️ Service '{self.name}' exceeding memory limit ({memory_mb:.0f}MB > {self.max_memory_mb}MB)")
                        self.log_buffer.append(f"[WARNING] Memory usage: {memory_mb:.0f}MB / {self.max_memory_mb}MB")
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    pass
                    
        except asyncio.CancelledError:
            pass
        except Exception as e:
            logger.error(f"Monitor error for '{self.name}': {str(e)}")
    
    async def _save_to_db(self):
        """Save service state to database"""
        try:
            async with aiosqlite.connect(str(settings.DB_PATH)) as db:
                await db.execute("""
                    INSERT OR REPLACE INTO services 
                    (id, name, type, language, entry_file, port, status, auto_restart, 
                     env_vars, app_dir, command, description, max_memory_mb, 
                     restart_count, last_started, last_stopped, pid, updated_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """, (
                    self.id, self.name, self.type, self.language, self.entry_file,
                    self.port, self.status, int(self.auto_restart),
                    json.dumps(self.env_vars) if isinstance(self.env_vars, dict) else self.env_vars,
                    self.app_dir, self.command, self.description, self.max_memory_mb,
                    self.restart_count,
                    self.last_started.isoformat() if self.last_started else None,
                    self.last_stopped.isoformat() if self.last_stopped else None,
                    self.pid, datetime.now().isoformat()
                ))
                await db.commit()
        except Exception as e:
            logger.error(f"DB save error: {str(e)}")
    
    def get_info(self) -> dict:
        """Get complete service information"""
        uptime = 0
        cpu_percent = 0
        memory_mb = 0
        
        if self.process and self.process.poll() is None and self.pid:
            try:
                proc = psutil.Process(self.pid)
                cpu_percent = proc.cpu_percent(interval=0.1)
                memory_mb = proc.memory_info().rss / (1024 * 1024)
                if self.uptime_start:
                    uptime = int(time.time() - self.uptime_start)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                pass
        
        return {
            "id": self.id,
            "name": self.name,
            "type": self.type,
            "language": self.language,
            "entry_file": self.entry_file,
            "command": self.command,
            "app_dir": self.app_dir,
            "port": self.port,
            "status": self.status,
            "pid": self.pid,
            "auto_restart": self.auto_restart,
            "restart_count": self.restart_count,
            "uptime_seconds": uptime,
            "uptime_formatted": self._format_uptime(uptime),
            "cpu_percent": round(cpu_percent, 1),
            "memory_mb": round(memory_mb, 1),
            "max_memory_mb": self.max_memory_mb,
            "last_started": self.last_started.isoformat() if self.last_started else None,
            "last_stopped": self.last_stopped.isoformat() if self.last_stopped else None,
            "env_vars": self.env_vars,
            "description": self.description,
            "recent_logs": list(self.log_buffer)[-50:]
        }
    
    @staticmethod
    def _format_uptime(seconds: int) -> str:
        """Format uptime in human readable format"""
        if seconds <= 0:
            return "0s"
        days = seconds // 86400
        hours = (seconds % 86400) // 3600
        minutes = (seconds % 3600) // 60
        secs = seconds % 60
        
        parts = []
        if days > 0:
            parts.append(f"{days}d")
        if hours > 0:
            parts.append(f"{hours}h")
        if minutes > 0:
            parts.append(f"{minutes}m")
        parts.append(f"{secs}s")
        return " ".join(parts)


class ProcessManager:
    """
    Central Process Manager - Manages all services
    This is the BRAIN of RUHI-CORE
    """
    
    def __init__(self):
        self.services: Dict[str, ServiceProcess] = {}
        self._initialized = False
        logger.info("🧠 ProcessManager initialized")
    
    async def initialize(self):
        """Load existing services from database"""
        if self._initialized:
            return
        
        try:
            await init_db_safe()
            
            async with aiosqlite.connect(str(settings.DB_PATH)) as db:
                db.row_factory = aiosqlite.Row
                async with db.execute("SELECT * FROM services") as cursor:
                    rows = await cursor.fetchall()
                    for row in rows:
                        service = ServiceProcess(
                            service_id=row["id"],
                            name=row["name"],
                            type=row["type"],
                            language=row["language"],
                            entry_file=row["entry_file"],
                            port=row["port"],
                            auto_restart=bool(row["auto_restart"]),
                            env_vars=row["env_vars"],
                            app_dir=row["app_dir"],
                            command=row["command"] or "",
                            description=row["description"] or "",
                            max_memory_mb=row["max_memory_mb"]
                        )
                        service.restart_count = row["restart_count"] or 0
                        self.services[service.id] = service
                        logger.info(f"📦 Loaded service: {service.name} ({service.type})")
            
            self._initialized = True
            logger.info(f"✅ ProcessManager loaded {len(self.services)} services from database")
            
        except Exception as e:
            logger.error(f"❌ ProcessManager initialization error: {str(e)}")
            self._initialized = True  # Mark as initialized to prevent infinite loops
    
    async def create_service(self, name: str, **kwargs) -> ServiceProcess:
        """Create a new service"""
        if len(self.services) >= settings.MAX_SERVICES:
            raise ValueError(f"Maximum services limit reached ({settings.MAX_SERVICES})")
        
        # Check duplicate name
        for svc in self.services.values():
            if svc.name == name:
                raise ValueError(f"Service '{name}' already exists")
        
        service_id = str(uuid.uuid4())[:12]
        app_dir = str(settings.APPS_DIR / name)
        os.makedirs(app_dir, exist_ok=True)
        
        service = ServiceProcess(
            service_id=service_id,
            name=name,
            app_dir=app_dir,
            **kwargs
        )
        
        self.services[service_id] = service
        await service._save_to_db()
        
        logger.info(f"✅ Service '{name}' created (ID: {service_id})")
        return service
    
    async def delete_service(self, service_id: str) -> bool:
        """Delete a service"""
        service = self.services.get(service_id)
        if not service:
            raise ValueError(f"Service {service_id} not found")
        
        # Stop if running
        if service.status == "running":
            await service.stop()
        
        # Release port
        port_manager.release(service_id)
        
        # Remove from memory
        del self.services[service_id]
        
        # Remove from database
        try:
            async with aiosqlite.connect(str(settings.DB_PATH)) as db:
                await db.execute("DELETE FROM services WHERE id = ?", (service_id,))
                await db.execute("DELETE FROM service_logs WHERE service_id = ?", (service_id,))
                await db.commit()
        except Exception as e:
            logger.error(f"DB delete error: {str(e)}")
        
        logger.info(f"🗑️ Service '{service.name}' deleted")
        return True
    
    async def start_service(self, service_id: str) -> bool:
        """Start a specific service"""
        service = self.services.get(service_id)
        if not service:
            raise ValueError(f"Service {service_id} not found")
        return await service.start()
    
    async def stop_service(self, service_id: str) -> bool:
        """Stop a specific service"""
        service = self.services.get(service_id)
        if not service:
            raise ValueError(f"Service {service_id} not found")
        return await service.stop()
    
    async def restart_service(self, service_id: str) -> bool:
        """Restart a specific service"""
        service = self.services.get(service_id)
        if not service:
            raise ValueError(f"Service {service_id} not found")
        return await service.restart()
    
    async def start_all(self):
        """Start all services that were previously running"""
        for service in self.services.values():
            if service.auto_restart:
                await service.start()
                await asyncio.sleep(1)  # Stagger starts
    
    async def stop_all(self):
        """Stop all running services"""
        for service in self.services.values():
            if service.status == "running":
                await service.stop()
    
    def get_service(self, service_id: str) -> Optional[ServiceProcess]:
        """Get a service by ID"""
        return self.services.get(service_id)
    
    def get_service_by_name(self, name: str) -> Optional[ServiceProcess]:
        """Get a service by name"""
        for service in self.services.values():
            if service.name == name:
                return service
        return None
    
    def get_all_services(self) -> List[dict]:
        """Get info of all services"""
        return [svc.get_info() for svc in self.services.values()]
    
    def get_stats(self) -> dict:
        """Get summary statistics"""
        running = sum(1 for s in self.services.values() if s.status == "running")
        stopped = sum(1 for s in self.services.values() if s.status == "stopped")
        crashed = sum(1 for s in self.services.values() if s.status == "crashed")
        
        return {
            "total_services": len(self.services),
            "running": running,
            "stopped": stopped,
            "crashed": crashed,
            "max_services": settings.MAX_SERVICES,
            "available_ports": port_manager.available_count,
            "port_allocations": port_manager.get_all_allocations()
        }


async def init_db_safe():
    """Safe database initialization"""
    try:
        from core.config import init_db
        await init_db()
    except Exception as e:
        logger.warning(f"DB init warning: {str(e)}")


# Global instance
process_manager = ProcessManager()