rspade_system/app/RSpade/Commands/Rsx/Task_Worker_Command.php

<?php
/**
 * CODING CONVENTION:
 * This file follows the coding convention where variable_names and function_names
 * use snake_case (underscore_wherever_possible).
 */

namespace App\RSpade\Commands\Rsx;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;
use App\RSpade\Core\Task\Task_Instance;
use App\RSpade\Core\Task\Task_Status;
use App\RSpade\Core\Task\Task_Lock;

/**
 * Task Worker Command
 *
 * Background worker process that continuously processes tasks from a queue.
 * Spawned by Task_Process_Command, runs until no more tasks or timeout.
 *
 * This command:
 * 1. Acquires lock for queue
 * 2. Selects next pending task (SELECT FOR UPDATE)
 * 3. Marks task as running
 * 4. Executes the task
 * 5. Marks task as completed/failed
 * 6. Repeats until no more tasks
 *
 * Exit conditions:
 * - No more pending tasks
 * - Maximum execution time reached (5 minutes)
 * - Fatal error
 */
class Task_Worker_Command extends Command
{
    protected $signature = 'rsx:task:worker
        {--queue=default : Queue name to process}
        {--max-time=300 : Maximum execution time in seconds (default: 5 minutes)}';

    protected $description = 'Background worker for processing queued tasks';

    private int $start_time;
    private int $max_time;
    private int $tasks_processed = 0;

    public function handle()
    {
        $queue_name = $this->option('queue');
        $this->max_time = (int) $this->option('max-time');
        $this->start_time = time();

        $this->info("[WORKER] Started worker for queue '{$queue_name}' (PID: " . getmypid() . ")");

        // Process tasks until none remain or timeout
        while (true) {
            // Check if we've exceeded max execution time
            if ((time() - $this->start_time) >= $this->max_time) {
                $this->info("[WORKER] Max execution time reached, exiting");
                break;
            }

            // Try to process one task
            $processed = $this->process_next_task($queue_name);

            if (!$processed) {
                // No more tasks to process
                $this->info("[WORKER] No more pending tasks, exiting");
                break;
            }

            $this->tasks_processed++;
        }

        $this->info("[WORKER] Worker finished. Processed {$this->tasks_processed} task(s)");

        return 0;
    }

    /**
     * Process the next pending task from the queue
     *
     * @param string $queue_name Queue to process from
     * @return bool True if task was processed, false if no tasks available
     */
    private function process_next_task(string $queue_name): bool
    {
        // Use lock to ensure atomic task selection
        $lock = new Task_Lock("task_queue_{$queue_name}", 5);

        if (!$lock->acquire()) {
            $this->warn("[WORKER] Could not acquire lock for queue '{$queue_name}', retrying...");
            sleep(1);
            return true; // Retry
        }

        try {
            // Select next pending task (exclude scheduled task records)
            $task_row = DB::table('_task_queue')
                ->where('queue', $queue_name)
                ->where('status', Task_Status::PENDING)
                ->whereNull('next_run_at')  // Exclude scheduled task records
                ->where(function ($query) {
                    $query->whereNull('scheduled_for')
                        ->orWhere('scheduled_for', '<=', now());
                })
                ->orderBy('created_at', 'asc')
                ->lockForUpdate()
                ->first();

            if (!$task_row) {
                // No tasks available
                $lock->release();
                return false;
            }

            // Mark as running
            DB::table('_task_queue')
                ->where('id', $task_row->id)
                ->update([
                    'status' => Task_Status::RUNNING,
                    'started_at' => now(),
                    'worker_pid' => getmypid(),
                    'updated_at' => now(),
                ]);

            $lock->release();

            // Execute the task
            $this->execute_task($task_row);

            return true;
        } catch (\Exception $e) {
            $this->error("[WORKER] Error processing task: " . $e->getMessage());

            if ($lock->is_locked()) {
                $lock->release();
            }

            return false;
        }
    }

    /**
     * Execute a task
     *
     * @param object $task_row Task database row
     */
    private function execute_task(object $task_row): void
    {
        $this->info("[WORKER] Executing task {$task_row->id}: {$task_row->class}::{$task_row->method}");

        $task_instance = Task_Instance::find($task_row->id);

        if (!$task_instance) {
            $this->error("[WORKER] Could not load task instance for task {$task_row->id}");
            return;
        }

        try {
            $class = $task_row->class;
            $method = $task_row->method;
            $params = json_decode($task_row->params, true) ?? [];

            // Check if class exists
            if (!class_exists($class)) {
                throw new \Exception("Class not found: {$class}");
            }

            // Check if method exists
            if (!method_exists($class, $method)) {
                throw new \Exception("Method not found: {$class}::{$method}");
            }

            // Execute the task method
            $result = $class::$method($task_instance, $params);

            // Mark as completed
            $task_instance->mark_completed($result);

            $this->info("[WORKER] Task {$task_row->id} completed successfully");
        } catch (\Exception $e) {
            // Mark as failed
            $task_instance->mark_failed($e->getMessage());

            $this->error("[WORKER] Task {$task_row->id} failed: " . $e->getMessage());
        }
    }
}