mp_audio_manage/app/Console/Commands/CheckAiTaskHealthCommand.php

<?php

namespace App\Console\Commands;

use App\Consts\BaseConst;
use App\Models\MpGeneratePicTask;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Redis;

/**
 * 图片任务健康检查 & 钉钉告警
 *
 * 由 app/Console/Kernel.php 每分钟调度一次（ai:check-task-health）。
 * 每轮采集下面 4 项指标，命中后需要"连续命中 N 次"才真正发出告警，避免瞬时波动误报；
 * 同一规则告警后进入冷却期，冷却期内不重复发送；指标恢复正常后发送一条"已恢复"通知。
 *
 * =========================== 通知规则一览（全部参数见下方常量区） ===========================
 *
 * 【规则A】队列积压（QUEUE_BACKLOG_*）
 *   判定：监控队列的 Redis 长度 >= QUEUE_BACKLOG_THRESHOLD 条，
 *         且连续 QUEUE_BACKLOG_CONSECUTIVE 次命中（每分钟检查一次）。
 *   含义：job 已经派发进队列但没人消费——worker 进程挂了/没启动，
 *         或队列名、Redis 实例与派发方不一致。
 *
 * 【规则B】任务长时间未启动（PENDING_*）
 *   判定：队列派发型模型（GPT-Image2、火山系列）中 status=pending
 *         且 created_at 早于 PENDING_TIMEOUT_MINUTES 分钟的任务数 > 0，
 *         且连续 PENDING_CONSECUTIVE 次命中。
 *   含义：任务已创建却一直没被 worker 取走，用户端会一直显示 0 进度。
 *   说明：即梦（jimeng_4.0）、NanoBanana 走内部排队/同步提交，等待属于正常，不参与本规则。
 *
 * 【规则C】任务长时间处理中（PROCESSING_*）
 *   判定：status=processing 且 started_at 早于 PROCESSING_TIMEOUT_MINUTES 分钟的任务数 > 0，
 *         且连续 PROCESSING_CONSECUTIVE 次命中。
 *   含义：任务已被 worker 取走但卡在第三方接口，或 worker 被强杀导致状态悬挂。
 *
 * 【规则D】失败突增（FAILED_*）
 *   判定：最近 FAILED_WINDOW_MINUTES 分钟内 status=failed 的任务数 > FAILED_THRESHOLD，
 *         且连续 FAILED_CONSECUTIVE 次命中。
 *   含义：第三方账号/额度异常（例如历史出现过的 USER_INACTIVE）或内容审核批量拦截。
 *
 * 【通用行为】
 *   1. 同一规则告警后，ALERT_COOLDOWN_MINUTES 分钟内不重复发送；超过冷却期仍异常则重复提醒一次；
 *   2. 指标恢复正常时发送一条"已恢复"通知（含本次异常持续时长）；
 *   3. 所有通知都带环境（APP_ENV）、命中详情与排查建议；
 *   4. 连续次数、冷却状态记录在 Redis（ALERT_KEY_PREFIX + 规则名），ALERT_STATE_TTL 后自动过期；
 *   5. 钉钉通知直接复用项目已有的 sendNotice()（读取 .env 的 DD_WEB_HOOK）；
 *      未配置 webhook 或发送失败都只记录日志，不影响命令执行。
 * ===========================================================================================
 */
class CheckAiTaskHealthCommand extends Command
{
    /**
     * 命令名与参数
     *
     * @var string
     */
    protected $signature = 'ai:check-task-health {--dry-run : 只输出判定结果，不发送钉钉也不写入状态}';

    /**
     * 命令说明
     *
     * @var string
     */
    protected $description = '图片任务健康检查（队列积压/任务卡住/失败突增）并在异常时发送钉钉告警';

    // ============================== 规则A：队列积压 ==============================
    /** 监控的 Redis 队列：队列名 => 展示名（Laravel Redis 队列的 key 为 queues:{队列名}） */
    const MONITOR_QUEUES = [
        '{GenerateGptImage2Pics}' => 'GPT-Image2',
        '{GenerateVolcPics}'      => '火山API',
    ];
    /** 队列积压条数阈值（达到即算命中） */
    const QUEUE_BACKLOG_THRESHOLD = 200;
    /** 队列积压需要连续命中的次数（每分钟检查一次） */
    const QUEUE_BACKLOG_CONSECUTIVE = 3;

    // ============================== 规则B：任务长时间未启动 ==============================
    /** pending 超时时间（分钟），超过即算命中 */
    const PENDING_TIMEOUT_MINUTES = 10;
    /** pending 超时需要连续命中的次数 */
    const PENDING_CONSECUTIVE = 2;

    // ============================== 规则C：任务长时间处理中 ==============================
    /** processing 超时时间（分钟），超过即算命中 */
    const PROCESSING_TIMEOUT_MINUTES = 15;
    /** processing 超时需要连续命中的次数 */
    const PROCESSING_CONSECUTIVE = 2;

    // ============================== 规则D：失败突增 ==============================
    /** 失败统计的滑动窗口（分钟） */
    const FAILED_WINDOW_MINUTES = 10;
    /** 窗口内失败条数阈值（超过才告警） */
    const FAILED_THRESHOLD = 20;
    /** 失败突增需要连续命中的次数 */
    const FAILED_CONSECUTIVE = 2;
    /** 失败样例在告警中展示的条数 */
    const FAILED_SAMPLE_LIMIT = 3;

    // ============================== 通用参数 ==============================
    /** 同一规则告警后的冷却时间（分钟），冷却期内不重复发送 */
    const ALERT_COOLDOWN_MINUTES = 30;
    /** 告警状态在 Redis 中的 key 前缀 */
    const ALERT_KEY_PREFIX = 'ai_task_alert:';
    /** 告警状态保留时间（秒），避免 Redis 中残留脏状态 */
    const ALERT_STATE_TTL = 604800; // 7 天
    /** 使用的 Redis 连接，必须与队列保持一致 */
    const ALERT_REDIS_CONNECTION = 'default';
    /** 展示错误信息时截断的字符数 */
    const ERROR_MESSAGE_LIMIT = 80;

    /**
     * 执行命令
     *
     * @return int
     */
    public function handle()
    {
        $dryRun = (bool) $this->option('dry-run');

        $this->info('图片任务健康检查开始' . ($dryRun ? '（dry-run：不发送钉钉、不写入状态）' : ''));
        $this->line('环境：' . config('app.env') . '　时间：' . now()->toDateTimeString());

        try {
            // 采集 4 项规则
            $rules = [
                $this->buildQueueBacklogRule(),
                $this->buildPendingTimeoutRule(),
                $this->buildProcessingTimeoutRule(),
                $this->buildFailedSpikeRule(),
            ];

            foreach ($rules as $rule) {
                $this->processRule($rule, $dryRun);
            }
        } catch (\Throwable $e) {
            // 检查本身异常不影响其它定时任务，只记录并输出
            $this->error('图片任务健康检查执行失败：' . $e->getMessage());
            dLog('generate')->error('图片任务健康检查执行失败', ['error' => $e->getMessage()]);

            return 1;
        }

        $this->info('图片任务健康检查结束');

        return 0;
    }

    /**
     * 规则处理：连续计数、冷却、告警与恢复通知、状态落库
     *
     * @param array $rule   规则数据
     * @param bool  $dryRun 是否演练模式（不发通知、不写状态）
     * @return void
     */
    private function processRule(array $rule, bool $dryRun): void
    {
        $state = $this->loadState($rule['key']);
        $cooldownSeconds = self::ALERT_COOLDOWN_MINUTES * 60;

        $this->outputRuleResult($rule, $state);

        if ($rule['triggered']) {
            $state['consecutive'] = (int) $state['consecutive'] + 1;

            if ($state['consecutive'] === 1 && empty($state['first_hit_at'])) {
                $state['first_hit_at'] = time();
            }

            if (!$state['alerting']) {
                // 未处于告警状态：连续命中达到配置次数才发送
                if ($state['consecutive'] >= (int) $rule['consecutive']) {
                    if ($dryRun) {
                        $this->line('  → [dry-run] 已达到连续命中次数，将发送告警');
                    } else {
                        $this->sendAlert($rule, false);
                        $state['alerting'] = true;
                        $state['alerted_at'] = time();
                    }
                } else {
                    $this->line('  → 命中 ' . $state['consecutive'] . '/' . $rule['consecutive'] . ' 次，未达连续次数，暂不告警');
                }
            } else {
                // 已在告警中：冷却结束后重复提醒
                $elapsed = time() - (int) $state['alerted_at'];
                if ($elapsed >= $cooldownSeconds) {
                    if ($dryRun) {
                        $this->line('  → [dry-run] 冷却期已过且异常仍在持续，将重复发送告警');
                    } else {
                        $this->sendAlert($rule, true);
                        $state['alerted_at'] = time();
                    }
                } else {
                    $this->line('  → 告警中，冷却剩余约 ' . (int) ceil(($cooldownSeconds - $elapsed) / 60) . ' 分钟');
                }
            }
        } else {
            if ($state['alerting']) {
                if ($dryRun) {
                    $this->line('  → [dry-run] 指标已恢复，将发送恢复通知');
                } else {
                    $this->sendRecovery($rule, $state);
                }
                $state['alerting'] = false;
            }
            $state['consecutive'] = 0;
            $state['first_hit_at'] = 0;
        }

        if (!$dryRun) {
            $state['last_detail'] = implode(' | ', $rule['detail']);
            $state['last_check_at'] = time();
            $this->saveState($rule['key'], $state);
        }
    }

    /**
     * 规则A：队列积压
     *
     * @return array
     */
    private function buildQueueBacklogRule(): array
    {
        $detail = [];
        $hit = [];

        try {
            foreach (self::MONITOR_QUEUES as $queue => $label) {
                $length = (int) Redis::connection(self::ALERT_REDIS_CONNECTION)->llen('queues:' . $queue);
                $detail[] = $label . '队列（queues:' . $queue . '）：' . $length . ' 条，阈值 ' . self::QUEUE_BACKLOG_THRESHOLD . ' 条';

                if ($length >= self::QUEUE_BACKLOG_THRESHOLD) {
                    $hit[] = $label . '=' . $length;
                }
            }
        } catch (\Throwable $e) {
            // Redis 不可用时无法判定队列长度，只提示不告警（避免误报）
            $detail[] = 'Redis 队列长度读取失败：' . $e->getMessage();
            $hit = [];
        }

        return [
            'key'         => 'queue_backlog',
            'title'       => '队列积压',
            'rule_text'   => '队列长度 >= ' . self::QUEUE_BACKLOG_THRESHOLD . ' 条，连续 ' . self::QUEUE_BACKLOG_CONSECUTIVE . ' 次命中',
            'consecutive' => self::QUEUE_BACKLOG_CONSECUTIVE,
            'triggered'   => !empty($hit),
            'detail'      => $detail,
            'hit_text'    => implode('，', $hit),
            'suggest'     => '检查 worker 进程是否存活（supervisorctl status generate_gpt_image2_pics:*），确认队列名与 Redis 实例和派发方一致',
        ];
    }

    /**
     * 规则B：任务长时间未启动
     *
     * @return array
     */
    private function buildPendingTimeoutRule(): array
    {
        $deadline = now()->subMinutes(self::PENDING_TIMEOUT_MINUTES);
        $query = DB::table('mp_generate_pic_tasks')
            ->where('status', MpGeneratePicTask::STATUS_PENDING)
            ->whereIn('model', $this->pendingWatchModels())
            ->where('created_at', '<', $deadline);

        $count = (clone $query)->count();
        $detail = ['pending 超过 ' . self::PENDING_TIMEOUT_MINUTES . ' 分钟的任务：' . $count . ' 条'];

        if ($count > 0) {
            $oldest = (clone $query)->orderBy('created_at', 'asc')->first();
            if ($oldest) {
                $detail[] = '最早一条：id=' . $oldest->id
                    . '，模型=' . $oldest->model
                    . '，外部任务号=' . $oldest->task_id
                    . '，创建于 ' . $oldest->created_at
                    . '（已等待 ' . $this->minutesSince($oldest->created_at) . ' 分钟）';
            }
        }

        return [
            'key'         => 'pending_timeout',
            'title'       => '任务长时间未启动',
            'rule_text'   => 'pending 超过 ' . self::PENDING_TIMEOUT_MINUTES . ' 分钟的任务数 > 0，连续 ' . self::PENDING_CONSECUTIVE . ' 次命中',
            'consecutive' => self::PENDING_CONSECUTIVE,
            'triggered'   => $count > 0,
            'detail'      => $detail,
            'hit_text'    => '超时 pending ' . $count . ' 条',
            'suggest'     => '任务已入队但未被消费，优先检查对应队列的 worker 与队列积压情况',
        ];
    }

    /**
     * 规则C：任务长时间处理中
     *
     * @return array
     */
    private function buildProcessingTimeoutRule(): array
    {
        $deadline = now()->subMinutes(self::PROCESSING_TIMEOUT_MINUTES);
        $query = DB::table('mp_generate_pic_tasks')
            ->where('status', MpGeneratePicTask::STATUS_PROCESSING)
            ->whereNotNull('started_at')
            ->where('started_at', '<', $deadline);

        $count = (clone $query)->count();
        $detail = ['processing 超过 ' . self::PROCESSING_TIMEOUT_MINUTES . ' 分钟的任务：' . $count . ' 条'];

        if ($count > 0) {
            $oldest = (clone $query)->orderBy('started_at', 'asc')->first();
            if ($oldest) {
                $detail[] = '最早一条：id=' . $oldest->id
                    . '，模型=' . $oldest->model
                    . '，外部任务号=' . $oldest->task_id
                    . '，开始于 ' . $oldest->started_at
                    . '（已处理 ' . $this->minutesSince($oldest->started_at) . ' 分钟）';
            }
        }

        return [
            'key'         => 'processing_timeout',
            'title'       => '任务长时间处理中',
            'rule_text'   => 'processing 超过 ' . self::PROCESSING_TIMEOUT_MINUTES . ' 分钟的任务数 > 0，连续 ' . self::PROCESSING_CONSECUTIVE . ' 次命中',
            'consecutive' => self::PROCESSING_CONSECUTIVE,
            'triggered'   => $count > 0,
            'detail'      => $detail,
            'hit_text'    => '超时 processing ' . $count . ' 条',
            'suggest'     => '检查第三方接口是否超时/限流，以及 worker 是否被强杀（supervisor 的 stopwaitsecs 是否小于任务超时时间）',
        ];
    }

    /**
     * 规则D：失败突增
     *
     * @return array
     */
    private function buildFailedSpikeRule(): array
    {
        $deadline = now()->subMinutes(self::FAILED_WINDOW_MINUTES);
        $query = DB::table('mp_generate_pic_tasks')
            ->where('status', MpGeneratePicTask::STATUS_FAILED)
            ->where('updated_at', '>=', $deadline);

        $count = (clone $query)->count();
        $detail = ['最近 ' . self::FAILED_WINDOW_MINUTES . ' 分钟内失败的任务：' . $count . ' 条，阈值 > ' . self::FAILED_THRESHOLD . ' 条'];

        if ($count > self::FAILED_THRESHOLD) {
            $samples = (clone $query)->orderBy('updated_at', 'desc')->limit(self::FAILED_SAMPLE_LIMIT)->get(['id', 'model', 'error_message', 'updated_at']);
            foreach ($samples as $sample) {
                $detail[] = '样例：id=' . $sample->id
                    . '，模型=' . $sample->model
                    . '，时间=' . $sample->updated_at
                    . '，错误=' . mb_substr((string) $sample->error_message, 0, self::ERROR_MESSAGE_LIMIT);
            }
        }

        return [
            'key'         => 'failed_spike',
            'title'       => '失败任务突增',
            'rule_text'   => '最近 ' . self::FAILED_WINDOW_MINUTES . ' 分钟内失败数 > ' . self::FAILED_THRESHOLD . ' 条，连续 ' . self::FAILED_CONSECUTIVE . ' 次命中',
            'consecutive' => self::FAILED_CONSECUTIVE,
            'triggered'   => $count > self::FAILED_THRESHOLD,
            'detail'      => $detail,
            'hit_text'    => '失败 ' . $count . ' 条',
            'suggest'     => '检查第三方账号/额度状态与内容审核拦截，确认接口返回的错误码',
        ];
    }

    /**
     * pending 规则监控的模型：仅"队列派发型"（GPT-Image2、火山系列）
     *
     * 即梦（jimeng_4.0）、NanoBanana 走内部排队/同步提交，长时间 pending 属正常，不参与告警。
     *
     * @return array
     */
    private function pendingWatchModels(): array
    {
        return array_merge(BaseConst::GPT_IMAGE2_MODELS, BaseConst::VOLC_PIC_MODELS);
    }

    /**
     * 控制台输出单条规则的判定结果
     *
     * @param array $rule
     * @param array $state
     * @return void
     */
    private function outputRuleResult(array $rule, array $state): void
    {
        $status = $rule['triggered']
            ? '命中（连续 ' . ((int) $state['consecutive'] + 1) . '/' . $rule['consecutive'] . '）'
            : '正常';

        $this->line('【' . $rule['title'] . '】' . $status);

        foreach ($rule['detail'] as $line) {
            $this->line('  - ' . $line);
        }
    }

    /**
     * 发送告警（钉钉 + 控制台 + 日志）
     *
     * @param array $rule
     * @param bool  $repeat 是否为持续告警的重复提醒
     * @return void
     */
    private function sendAlert(array $rule, bool $repeat): void
    {
        $lines = [
            ($repeat ? '【图片任务告警·持续】' : '【图片任务告警】') . $rule['title'],
            '环境：' . config('app.env'),
            '规则：' . $rule['rule_text'],
            '当前：' . $rule['hit_text'],
        ];

        foreach ($rule['detail'] as $line) {
            $lines[] = '  - ' . $line;
        }

        if ($repeat) {
            $lines[] = '说明：该异常仍在持续，距上次通知已 ' . self::ALERT_COOLDOWN_MINUTES . ' 分钟';
        }

        $lines[] = '建议：' . $rule['suggest'];
        $lines[] = '时间：' . now()->toDateTimeString();

        $content = implode("\n", $lines);

        $this->error($content);
        dLog('generate')->error('图片任务健康告警：' . $rule['title'], [
            'rule'   => $rule['key'],
            'detail' => $rule['detail'],
        ]);
        logDB('generate', 'error', '图片任务健康告警：' . $rule['title'], [
            'rule'   => $rule['key'],
            'detail' => $rule['detail'],
        ]);

        $this->notify($content);
    }

    /**
     * 发送恢复通知（只在上一轮处于告警状态时发送）
     *
     * @param array $rule
     * @param array $state
     * @return void
     */
    private function sendRecovery(array $rule, array $state): void
    {
        $lines = [
            '【图片任务恢复】' . $rule['title'] . ' 已恢复正常',
            '环境：' . config('app.env'),
        ];

        if (!empty($state['first_hit_at'])) {
            $lines[] = '本次异常持续约 ' . (int) ceil((time() - (int) $state['first_hit_at']) / 60) . ' 分钟';
        }

        foreach ($rule['detail'] as $line) {
            $lines[] = '  - ' . $line;
        }

        $lines[] = '时间：' . now()->toDateTimeString();

        $content = implode("\n", $lines);

        $this->info($content);
        dLog('generate')->info('图片任务健康恢复：' . $rule['title'], ['rule' => $rule['key']]);
        logDB('generate', 'info', '图片任务健康恢复：' . $rule['title'], ['rule' => $rule['key']]);

        $this->notify($content);
    }

    /**
     * 发送钉钉通知
     *
     * 直接复用项目已有的 sendNotice()（读取线上 .env 的 DD_WEB_HOOK，默认 @所有人）。
     * webhook 未配置或发送失败都只记录日志，绝不影响检查命令本身。
     *
     * @param string $content
     * @return void
     */
    private function notify(string $content): void
    {
        if (empty(env('DD_WEB_HOOK'))) {
            $this->warn('DD_WEB_HOOK 未配置，跳过钉钉发送');
            dLog('generate')->warning('图片任务健康告警未发送：DD_WEB_HOOK 未配置', ['content' => $content]);
            return;
        }

        try {
            sendNotice($content);
        } catch (\Throwable $e) {
            $this->warn('钉钉发送失败：' . $e->getMessage());
            dLog('generate')->error('图片任务健康告警发送失败', [
                'error'   => $e->getMessage(),
                'content' => $content,
            ]);
        }
    }

    /**
     * 读取规则状态（连续命中次数、是否告警中、首次命中时间等）
     *
     * @param string $ruleKey
     * @return array
     */
    private function loadState(string $ruleKey): array
    {
        $default = [
            'consecutive' => 0,
            'alerting'    => false,
            'alerted_at'  => 0,
            'first_hit_at'=> 0,
        ];

        try {
            $raw = Redis::connection(self::ALERT_REDIS_CONNECTION)->get($this->stateKey($ruleKey));
            if (empty($raw)) {
                return $default;
            }

            $state = json_decode($raw, true);
            if (!is_array($state)) {
                return $default;
            }

            return array_merge($default, $state);
        } catch (\Throwable $e) {
            dLog('generate')->warning('读取图片任务告警状态失败', [
                'rule'  => $ruleKey,
                'error' => $e->getMessage(),
            ]);

            return $default;
        }
    }

    /**
     * 保存规则状态
     *
     * @param string $ruleKey
     * @param array  $state
     * @return void
     */
    private function saveState(string $ruleKey, array $state): void
    {
        try {
            Redis::connection(self::ALERT_REDIS_CONNECTION)->set(
                $this->stateKey($ruleKey),
                json_encode($state, JSON_UNESCAPED_UNICODE),
                'EX',
                self::ALERT_STATE_TTL
            );
        } catch (\Throwable $e) {
            dLog('generate')->warning('保存图片任务告警状态失败', [
                'rule'  => $ruleKey,
                'error' => $e->getMessage(),
            ]);
        }
    }

    /**
     * 规则状态在 Redis 中的 key
     *
     * @param string $ruleKey
     * @return string
     */
    private function stateKey(string $ruleKey): string
    {
        return self::ALERT_KEY_PREFIX . 'state:' . $ruleKey;
    }

    /**
     * 计算某个时间点距今经过的分钟数
     *
     * @param string $dateTime
     * @return int
     */
    private function minutesSince(string $dateTime): int
    {
        $timestamp = strtotime($dateTime);

        return $timestamp ? (int) floor((time() - $timestamp) / 60) : 0;
    }
}