easystream-main/start-workers.php

<?php
/*******************************************************************************************************************
| EasyStream Queue Workers Starter
| Starts background workers for video processing and system maintenance
|*******************************************************************************************************************/

define('_ISVALID', true);

// Include core configuration
include_once 'f_core/config.core.php';

/**
 * Queue Workers Management Script
 */
class WorkerManager
{
    private $queueManager;
    private $logger;
    private $workers = [];
    
    public function __construct()
    {
        $this->queueManager = new VQueueManager();
        $this->logger = VLogger::getInstance();
    }
    
    /**
     * Start all workers
     */
    public function startWorkers()
    {
        echo "Starting EasyStream Queue Workers...\n";
        echo str_repeat("=", 50) . "\n";
        
        // Start video processing workers
        $this->startVideoWorkers(2);
        
        // Start general workers
        $this->startGeneralWorkers(1);
        
        // Schedule cleanup jobs
        $this->scheduleCleanupJobs();
        
        echo "\nAll workers started successfully!\n";
        echo "Press Ctrl+C to stop all workers\n\n";
        
        // Keep script running
        $this->monitorWorkers();
    }
    
    /**
     * Start video processing workers
     * @param int $count Number of workers to start
     */
    private function startVideoWorkers($count)
    {
        echo "Starting {$count} video processing workers...\n";
        
        for ($i = 1; $i <= $count; $i++) {
            $cmd = "php f_core/f_workers/queue-worker.php video_processing > logs/video_worker_{$i}.log 2>&1 &";
            
            if (PHP_OS_FAMILY === 'Windows') {
                $cmd = "start /B php f_core/f_workers/queue-worker.php video_processing > logs/video_worker_{$i}.log 2>&1";
            }
            
            $pid = $this->executeCommand($cmd);
            
            if ($pid) {
                $this->workers[] = [
                    'type' => 'video_processing',
                    'id' => $i,
                    'pid' => $pid,
                    'started' => time()
                ];
                
                echo "  ✓ Video worker {$i} started (PID: {$pid})\n";
            } else {
                echo "  ✗ Failed to start video worker {$i}\n";
            }
        }
    }
    
    /**
     * Start general workers
     * @param int $count Number of workers to start
     */
    private function startGeneralWorkers($count)
    {
        echo "Starting {$count} general workers...\n";
        
        for ($i = 1; $i <= $count; $i++) {
            $cmd = "php f_core/f_workers/queue-worker.php default notifications > logs/general_worker_{$i}.log 2>&1 &";
            
            if (PHP_OS_FAMILY === 'Windows') {
                $cmd = "start /B php f_core/f_workers/queue-worker.php default notifications > logs/general_worker_{$i}.log 2>&1";
            }
            
            $pid = $this->executeCommand($cmd);
            
            if ($pid) {
                $this->workers[] = [
                    'type' => 'general',
                    'id' => $i,
                    'pid' => $pid,
                    'started' => time()
                ];
                
                echo "  ✓ General worker {$i} started (PID: {$pid})\n";
            } else {
                echo "  ✗ Failed to start general worker {$i}\n";
            }
        }
    }
    
    /**
     * Schedule cleanup jobs
     */
    private function scheduleCleanupJobs()
    {
        echo "Scheduling cleanup jobs...\n";
        
        // Schedule daily cleanup at 2 AM
        $cleanupData = [
            'type' => 'full',
            'log_retention_days' => 30,
            'temp_file_max_age_hours' => 24,
            'session_max_age_hours' => 168,
            'failed_upload_max_age_hours' => 48,
            'analytics_retention_days' => 90,
            'queue_retention_hours' => 24
        ];
        
        // Calculate delay until 2 AM tomorrow
        $now = time();
        $tomorrow2AM = strtotime('tomorrow 2:00 AM');
        $delay = $tomorrow2AM - $now;
        
        $jobId = $this->queueManager->enqueue('CleanupJob', $cleanupData, 'maintenance', $delay, 1);
        
        if ($jobId) {
            echo "  ✓ Daily cleanup job scheduled (Job ID: {$jobId})\n";
            echo "    Next run: " . date('Y-m-d H:i:s', $tomorrow2AM) . "\n";
        } else {
            echo "  ✗ Failed to schedule cleanup job\n";
        }
        
        // Schedule weekly database optimization
        $dbOptimizeData = [
            'type' => 'database'
        ];
        
        // Calculate delay until Sunday 3 AM
        $nextSunday3AM = strtotime('next Sunday 3:00 AM');
        $weeklyDelay = $nextSunday3AM - $now;
        
        $weeklyJobId = $this->queueManager->enqueue('CleanupJob', $dbOptimizeData, 'maintenance', $weeklyDelay, 1);
        
        if ($weeklyJobId) {
            echo "  ✓ Weekly database optimization scheduled (Job ID: {$weeklyJobId})\n";
            echo "    Next run: " . date('Y-m-d H:i:s', $nextSunday3AM) . "\n";
        }
    }
    
    /**
     * Monitor workers and restart if needed
     */
    private function monitorWorkers()
    {
        while (true) {
            sleep(30); // Check every 30 seconds
            
            foreach ($this->workers as $index => $worker) {
                if (!$this->isProcessRunning($worker['pid'])) {
                    echo "Worker {$worker['type']}_{$worker['id']} (PID: {$worker['pid']}) has stopped. Restarting...\n";
                    
                    // Restart the worker
                    $this->restartWorker($worker, $index);
                }
            }
            
            // Log worker status every 5 minutes
            if (time() % 300 === 0) {
                $this->logWorkerStatus();
            }
        }
    }
    
    /**
     * Restart a worker
     * @param array $worker Worker info
     * @param int $index Worker index
     */
    private function restartWorker($worker, $index)
    {
        if ($worker['type'] === 'video_processing') {
            $cmd = "php f_core/f_workers/queue-worker.php video_processing > logs/video_worker_{$worker['id']}.log 2>&1 &";
        } else {
            $cmd = "php f_core/f_workers/queue-worker.php default notifications > logs/general_worker_{$worker['id']}.log 2>&1 &";
        }
        
        if (PHP_OS_FAMILY === 'Windows') {
            $cmd = str_replace(' &', '', $cmd);
            $cmd = "start /B " . $cmd;
        }
        
        $pid = $this->executeCommand($cmd);
        
        if ($pid) {
            $this->workers[$index]['pid'] = $pid;
            $this->workers[$index]['started'] = time();
            echo "  ✓ Worker restarted with PID: {$pid}\n";
        } else {
            echo "  ✗ Failed to restart worker\n";
        }
    }
    
    /**
     * Execute command and return PID
     * @param string $cmd Command to execute
     * @return int|false PID or false on failure
     */
    private function executeCommand($cmd)
    {
        if (PHP_OS_FAMILY === 'Windows') {
            pclose(popen($cmd, 'r'));
            return true; // Windows doesn't easily return PID
        } else {
            $output = shell_exec($cmd . ' echo $!');
            return $output ? (int)trim($output) : false;
        }
    }
    
    /**
     * Check if process is running
     * @param int $pid Process ID
     * @return bool True if running
     */
    private function isProcessRunning($pid)
    {
        if (PHP_OS_FAMILY === 'Windows') {
            $output = shell_exec("tasklist /FI \"PID eq {$pid}\" 2>NUL");
            return strpos($output, (string)$pid) !== false;
        } else {
            return file_exists("/proc/{$pid}");
        }
    }
    
    /**
     * Log worker status
     */
    private function logWorkerStatus()
    {
        $runningWorkers = count($this->workers);
        $queueStats = $this->queueManager->getQueueStatistics();
        
        $this->logger->info('Worker status check', [
            'running_workers' => $runningWorkers,
            'queue_stats' => $queueStats,
            'uptime' => time() - $this->workers[0]['started']
        ]);
    }
    
    /**
     * Stop all workers
     */
    public function stopWorkers()
    {
        echo "Stopping all workers...\n";
        
        foreach ($this->workers as $worker) {
            if (PHP_OS_FAMILY === 'Windows') {
                shell_exec("taskkill /PID {$worker['pid']} /F 2>NUL");
            } else {
                shell_exec("kill {$worker['pid']} 2>/dev/null");
            }
            
            echo "  ✓ Stopped worker {$worker['type']}_{$worker['id']} (PID: {$worker['pid']})\n";
        }
        
        echo "All workers stopped.\n";
    }
    
    /**
     * Show help
     */
    public function showHelp()
    {
        echo "\nEasyStream Queue Workers Manager\n";
        echo str_repeat("=", 40) . "\n\n";
        echo "Usage: php start-workers.php [command]\n\n";
        echo "Commands:\n";
        echo "  start     Start all workers (default)\n";
        echo "  stop      Stop all workers\n";
        echo "  status    Show worker status\n";
        echo "  help      Show this help\n\n";
        echo "Examples:\n";
        echo "  php start-workers.php\n";
        echo "  php start-workers.php start\n";
        echo "  php start-workers.php stop\n\n";
    }
    
    /**
     * Show worker status
     */
    public function showStatus()
    {
        echo "EasyStream Workers Status\n";
        echo str_repeat("=", 30) . "\n\n";
        
        $queueStats = $this->queueManager->getQueueStatistics();
        $health = $this->queueManager->healthCheck();
        
        echo "System Health: " . strtoupper($health['status']) . "\n";
        echo "Backend: " . $health['backend'] . "\n\n";
        
        if (!empty($queueStats)) {
            echo "Queue Statistics:\n";
            foreach ($queueStats as $queue => $stats) {
                echo "  {$queue}:\n";
                echo "    Pending: " . ($stats['pending'] ?? 0) . "\n";
                echo "    Processing: " . ($stats['processing'] ?? 0) . "\n";
                echo "    Completed: " . ($stats['completed'] ?? 0) . "\n";
                echo "    Failed: " . ($stats['failed'] ?? 0) . "\n\n";
            }
        }
    }
}

// CLI execution
if (php_sapi_name() === 'cli') {
    $manager = new WorkerManager();
    
    $command = $argv[1] ?? 'start';
    
    // Handle shutdown signals
    if (function_exists('pcntl_signal')) {
        pcntl_signal(SIGTERM, function() use ($manager) {
            $manager->stopWorkers();
            exit(0);
        });
        pcntl_signal(SIGINT, function() use ($manager) {
            $manager->stopWorkers();
            exit(0);
        });
    }
    
    switch ($command) {
        case 'start':
            $manager->startWorkers();
            break;
            
        case 'stop':
            $manager->stopWorkers();
            break;
            
        case 'status':
            $manager->showStatus();
            break;
            
        case 'help':
        case '--help':
        case '-h':
            $manager->showHelp();
            break;
            
        default:
            echo "Unknown command: {$command}\n";
            $manager->showHelp();
            exit(1);
    }
}
?>