<?php

declare(strict_types=1);

namespace app\command;

use app\common\service\prescriptionai\PrescriptionAiStore;
use app\common\service\prescriptionai\PrescriptionAiWorker;
use think\console\Command;
use think\facade\Config;
use think\facade\Db;
use think\console\Input;
use think\console\input\Option;
use think\console\Output;

final class PrescriptionAiWork extends Command
{
    protected function configure()
    {
        $this->setName('prescription-ai:work')->setDescription('处方AI独立后台任务；分别运行prepare/qwen/openai')
            ->addOption('lane', null, Option::VALUE_REQUIRED, 'prepare、qwen、openai', 'prepare')
            ->addOption('once', null, Option::VALUE_NONE, '只处理一轮');
    }

    protected function execute(Input $input, Output $output): int
    {
        $lane = (string) $input->getOption('lane');
        if (!in_array($lane, ['prepare', 'qwen', 'openai'], true)) {
            $output->writeln('Invalid lane');
            return 1;
        }
        $running = true;
        if (function_exists('pcntl_async_signals')) {
            pcntl_async_signals(true);
            pcntl_signal(SIGTERM, static function () use (&$running): void { $running = false; });
            pcntl_signal(SIGINT, static function () use (&$running): void { $running = false; });
        }
        // One model task holds its database connection across several minutes of upstream calls,
        // which can outlive the server's wait_timeout. Without reconnecting, the first dropped
        // connection would wedge this consumer in a permanent error loop.
        $database = (array) config('database');
        $connection = (string) ($database['default'] ?? 'mysql');
        if (isset($database['connections'][$connection]) && is_array($database['connections'][$connection])) {
            $database['connections'][$connection]['break_reconnect'] = true;
            Config::set($database, 'database');
        }
        $worker = new PrescriptionAiWorker();
        $sweepAt = 0;
        $sourceCursor = 0;
        $rxCursor = 0;
        do {
            $worked = false;
            try {
                if (PrescriptionAiStore::enabled()) {
                    $worked = $lane === 'prepare' ? $worker->prepareOne() : $worker->runOne($lane);
                    if ($lane === 'prepare' && time() >= $sweepAt) {
                        $sweep = $worker->refreshSources($sourceCursor);
                        $sourceCursor = $sweep['selected'] > 0 ? $sweep['last_id'] : 0;
                        $rx = $worker->reconcile($rxCursor);
                        $rxCursor = $rx['selected'] > 0 ? $rx['last_id'] : 0;
                        $sweepAt = time() + 60;
                    }
                }
                if ($input->getOption('once') || $worked) {
                    $output->writeln('PRESCRIPTION_AI ' . json_encode(['lane' => $lane, 'enabled' => PrescriptionAiStore::enabled(), 'processed' => $worked]));
                }
            } catch (\Throwable $e) {
                // Class, location and SQLSTATE only: an exception message can carry SQL values or clinical text.
                $detail = get_class($e) . '@' . basename($e->getFile()) . ':' . $e->getLine();
                if (preg_match('/SQLSTATE\[[A-Z0-9]{5}\](?:\s*\[\d+\])?/', $e->getMessage(), $sqlState) === 1) {
                    $detail .= ' ' . $sqlState[0];
                }
                $output->writeln('PRESCRIPTION_AI storage_or_configuration_error ' . $detail);
                // Drop a possibly dead connection so the next round reconnects instead of looping.
                try {
                    Db::connect()->close();
                } catch (\Throwable $ignored) {
                }
                if ($input->getOption('once')) {
                    return 1;
                }
            }
            if (!$input->getOption('once') && $running) {
                usleep($worked ? 100000 : 1000000);
            }
        } while (!$input->getOption('once') && $running);
        return 0;
    }
}