mp_audio_manage/app/Console/Commands/TextScriptGenerateDispatchCommand.php

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;

/**
 * 剧本资产生成异步任务调度器
 *
 * 由 cron 每分钟触发一次。进入后以 5 秒为间隔轮询 pending 任务，
 * 发现任务立即拉起独立 worker 子进程（text:generate:run）；
 * 若连续 50 秒没有任何任务则退出，避免长时间占用调度进程。
 */
class TextScriptGenerateDispatchCommand extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'text:generate:dispatch';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = '剧本资产生成异步任务调度：每5秒查询pending任务并拉起worker';

    /**
     * Execute the console command.
     *
     * @return int
     */
    public function handle()
    {
        dLog('command')->info('开始执行剧本资产生成任务调度...');

        // 每次调度窗口执行一次历史 chunks 清理（任务完结后 chunks 仅供 SSE 推送，无长期保留价值）
        $cleaned = $this->cleanupExpiredChunks();
        if ($cleaned > 0) {
            dLog('command')->info("已清理过期剧本资产生成chunks: {$cleaned} 条");
        }

        $concurrency = (int)env('SCRIPT_TEXT_GENERATION_CONCURRENCY', 3);
        if ($concurrency < 1) {
            $concurrency = 1;
        }

        $start = time();
        $idleSince = null;
        $spawnedTotal = 0;

        while (true) {
            $now = time();

            // 孤儿任务恢复：processing 但超过5分钟且处理锁已空闲，说明 worker 异常退出，重置回 pending
            $this->recoverOrphanTasks();

            // 连续 50 秒没有任何任务则退出
            if ($idleSince !== null && $now - $idleSince >= 50) {
                dLog('command')->info('已连续50秒无待处理任务，退出本次调度');
                break;
            }

            // 整体安全上限（防止异常情况下长时间占用）
            if ($now - $start > 75) {
                dLog('command')->info('达到调度窗口安全上限，退出本次调度');
                break;
            }

            // 并发上限 = 配置上限 - 已在运行的 processing 任务数 - 本窗口已拉起数
            $activeCount = (int)DB::table('mp_script_generate_tasks')
                ->where('status', 'processing')
                ->count();
            $available = $concurrency - $activeCount - $spawnedTotal;
            if ($available <= 0) {
                dLog('command')->info("当前活跃任务已达并发上限，等待worker完成...");
                sleep(5);
                continue;
            }

            $pendingIds = DB::table('mp_script_generate_tasks')
                ->where('status', 'pending')
                ->orderBy('id')
                ->pluck('id')
                ->all();

            if (empty($pendingIds)) {
                if ($idleSince === null) {
                    $idleSince = $now;
                }
                if ($now - $idleSince >= 50) {
                    break;
                }
                sleep(5);
                continue;
            }

            $idleSince = null;
            foreach ($pendingIds as $taskId) {
                if ($available <= 0) {
                    break;
                }

                // 仅当该任务的处理锁空闲时才拉起，避免与手动/其它调度重复
                $lockName = TextScriptGenerateRunCommand::LOCK_PREFIX . $taskId;
                $isFree = DB::selectOne('SELECT IS_FREE_LOCK(?) AS is_free', [$lockName]);
                if (!$isFree || (int)$isFree->is_free !== 1) {
                    dLog('command')->info("task_id: {$taskId} 正在被其他进程处理，跳过");
                    continue;
                }

                try {
                    $this->spawnWorker((int)$taskId);
                    $spawnedTotal++;
                    $available--;
                    dLog('command')->info("已拉起剧本资产生成worker, task_id: {$taskId}");
                } catch (\Throwable $e) {
                    dLog('command')->error("拉起剧本资产生成worker失败 task_id: {$taskId}: " . $e->getMessage());
                    logDB('command', 'error', '剧本资产生成拉起失败', [
                        'task_id' => (int)$taskId,
                        'error'   => $e->getMessage(),
                    ]);
                }
            }

            sleep(5);
        }

        dLog('command')->info("剧本资产生成任务调度结束，本次拉起worker数: {$spawnedTotal}");
        return 0;
    }

    /**
     * 清理已完结任务的过期 chunks
     *
     * 保留期默认 30 天，可通过环境变量 SCRIPT_GENERATE_CHUNK_RETENTION_DAYS 调整。
     * 只清理 status 为 success/failed 且 completed_at 早于保留期的任务；
     * 按任务 id 升序游标分批删除，避免单次删除量过大造成长时间锁表。
     *
     * @return int 删除的 chunks 记录数
     */
    private function cleanupExpiredChunks(): int
    {
        $retentionDays = (int)env('SCRIPT_GENERATE_CHUNK_RETENTION_DAYS', 7);
        if ($retentionDays < 1) {
            $retentionDays = 7;
        }

        $cutoff = date('Y-m-d H:i:s', strtotime("-{$retentionDays} days"));
        $batchRows = (int)env('SCRIPT_GENERATE_CHUNK_DELETE_BATCH', 5000);
        if ($batchRows < 1000) {
            $batchRows = 5000;
        }

        $deletedTotal = 0;

        // 按行数小批循环删除，避免单次 DELETE 量过大造成长事务/长锁。
        // MySQL 单表 DELETE 支持 LIMIT；目标表是 chunks，子查询引用 tasks，不同表无同表限制。
        $deleteSql = 'DELETE FROM mp_script_generate_task_chunks '
            . 'WHERE task_id IN ('
            . 'SELECT id FROM mp_script_generate_tasks '
            . 'WHERE status IN (?, ?) AND completed_at IS NOT NULL AND completed_at < ?'
            . ') LIMIT ' . (int)$batchRows;

        $maxRounds = 2000; // 防御性上限，正常数据远用不到
        for ($round = 0; $round < $maxRounds; $round++) {
            $deleted = DB::affectingStatement($deleteSql, ['success', 'failed', $cutoff]);
            if ($deleted <= 0) {
                break;
            }
            $deletedTotal += $deleted;
            if ($deleted < $batchRows) {
                break;
            }
        }

        return $deletedTotal;
    }

    /**
     * 拉起一个后台 worker 子进程处理指定任务
     *
     * @param int $taskId
     * @return void
     */
    private function spawnWorker(int $taskId)
    {
        $command = [
            $this->phpBinary(),
            base_path('artisan'),
            'text:generate:run',
            (string)$taskId,
        ];

        if (DIRECTORY_SEPARATOR === '/') {
            // Linux：用 nohup 后台启动并立即返回
            $logFile = storage_path('logs/text-generate-worker-' . date('Ymd') . '.log');
            $commandLine = 'nohup ' . implode(' ', array_map('escapeshellarg', $command))
                . ' >> ' . escapeshellarg($logFile) . ' 2>&1 &';
            exec($commandLine);
            dLog('command')->info('拉起剧本资产生成worker命令: ' . $commandLine);
        } else {
            // Windows：使用 cmd start /B 分离启动。
            // 注意：不能用 Symfony Process::start()，其对象析构时会 stop(0) 终止刚拉起的子进程。
            $logFile = storage_path('logs/text-generate-worker-' . date('Ymd') . '.log');
            $commandLine = 'start /B "" ' . implode(' ', array_map('escapeshellarg', $command))
                . ' >> ' . escapeshellarg($logFile) . ' 2>&1';
            pclose(popen($commandLine, 'r'));
            dLog('command')->info('拉起剧本资产生成worker命令: ' . $commandLine);
        }
    }

    /**
     * 恢复异常退出遗留的孤儿任务
     *
     * worker 进程被强杀/崩溃时不会执行收尾，任务会一直停留在 processing。
     * 若任务已 processing 超过 5 分钟且其处理锁处于空闲状态，说明原 worker 已不在，
     * 将任务重置回 pending，等待重新调度。
     *
     * @return int 恢复的任务数
     */
    private function recoverOrphanTasks(): int
    {
        $recovered = 0;
        $threshold = date('Y-m-d H:i:s', strtotime('-5 minutes'));

        $candidates = DB::table('mp_script_generate_tasks')
            ->where('status', 'processing')
            ->where('started_at', '<', $threshold)
            ->pluck('id')
            ->all();

        foreach ($candidates as $taskId) {
            $lockName = TextScriptGenerateRunCommand::LOCK_PREFIX . $taskId;
            $isFree = DB::selectOne('SELECT IS_FREE_LOCK(?) AS is_free', [$lockName]);
            if (!$isFree || (int)$isFree->is_free !== 1) {
                continue;
            }

            $updated = DB::table('mp_script_generate_tasks')
                ->where('id', $taskId)
                ->where('status', 'processing')
                ->update([
                    'status'        => 'pending',
                    'error_message' => null,
                    'completed_at'  => null,
                    'updated_at'    => date('Y-m-d H:i:s'),
                ]);
            if ($updated) {
                $recovered++;
                dLog('command')->warning("孤儿任务恢复, task_id: {$taskId} 已重置为pending");
                logDB('command', 'warning', '孤儿任务恢复', [
                    'task_id' => (int)$taskId,
                    'message' => 'worker异常退出，任务重置为pending等待重新调度',
                ]);
            }
        }

        return $recovered;
    }

    /**
     * 获取当前 PHP 可执行文件的真实路径
     *
     * @return string
     */
    private function phpBinary()
    {
        $envBinary = getenv('PHP_BINARY');
        if ($envBinary && is_executable($envBinary)) {
            return $envBinary;
        }

        if (DIRECTORY_SEPARATOR === '/' && ($real = @readlink('/proc/self/exe')) && is_executable($real)) {
            return $real;
        }

        return PHP_BINARY;
    }
}