links/patches/application/common/service/SplitTicketSyncDispatchService.php

<?php

declare(strict_types=1);

namespace app\common\service;

use app\common\library\scrm\AntiBotConfigBuilder;
use think\Db;

/**
 * 工单同步 CLI 后台投递
 *
 * Web / Cron 仅负责鉴权与投递，实际 Spider 在独立 CLI 进程中执行，
 * 避免长时间占用 PHP-FPM、Cron 全局锁或 Session 锁。
 *
 * 并行策略：
 * - API 优先类型（Whatshub 等）：独立并行投递，不占 Node Real Browser 槽位
 * - 直连 PHP 类型：独立并行投递
 * - Node 类型：不同 sessionKey（域名）并行；同一 sessionKey 内串行以命中 Browser 温池
 */
class SplitTicketSyncDispatchService
{
    /**
     * 投递手工同步任务到后台 CLI
     *
     * @param int[] $ticketIds 已通过权限校验的工单 ID
     * @return array{queued:int[],skipped:int[],failed:int[]}
     */
    public function dispatchManual(array $ticketIds): array
    {
        $lockService = new SplitTicketSyncLockService();
        $metaList = $this->loadTicketMetaList($ticketIds);

        /** @var list<array{id:int,ticket_type:string,ticket_url:string}> $toDispatch */
        $toDispatch = [];
        $skipped = [];
        $failed = [];

        foreach ($metaList as $meta) {
            $ticketId = (int) $meta['id'];
            if ((int) ($meta['manual_manage'] ?? 0) === 1) {
                $skipped[] = $ticketId;
                continue;
            }
            if ($lockService->isLocked($ticketId)) {
                $skipped[] = $ticketId;
                continue;
            }
            $toDispatch[] = $meta;
        }

        $dispatchResult = $this->dispatchTicketMetaList($toDispatch, 'manual');

        SplitTicketSyncLogger::log('web', 'manual sync dispatched', [
            'queued'  => $dispatchResult['queued'],
            'skipped' => $skipped,
            'failed'  => $dispatchResult['failed'],
            'phpCli'  => $this->resolvePhpBinary(),
        ]);

        return [
            'queued'  => $dispatchResult['queued'],
            'skipped' => $skipped,
            'failed'  => $dispatchResult['failed'],
        ];
    }

    /**
     * Cron 自动同步：将挑选后的到期工单投递到后台 CLI（Cron 进程立即返回）
     *
     * @param list<array{id:int,ticket_type:string,ticket_url:string}|Ticket|\think\Model> $tickets
     * @return array{
     *   queued:list<array<string,mixed>>,
     *   skipped:list<array<string,mixed>>,
     *   failed:list<array<string,mixed>>,
     *   stoppedByNodeBusy:bool
     * }
     */
    public function dispatchAuto(array $tickets): array
    {
        $lockService = new SplitTicketSyncLockService();
        /** @var list<array{id:int,ticket_type:string,ticket_url:string}> $toDispatch */
        $toDispatch = [];
        /** @var list<array<string,mixed>> $skipped */
        $skipped = [];

        foreach ($tickets as $ticket) {
            $meta = $this->normalizeTicketMeta($ticket);
            if ($meta === null) {
                continue;
            }
            $ticketId = (int) $meta['id'];
            if ($lockService->isLocked($ticketId)) {
                $skipped[] = $this->buildDispatchEntry(
                    $ticketId,
                    (string) $meta['ticket_type'],
                    (string) $meta['ticket_url'],
                    '工单正在同步中'
                );
                continue;
            }
            $toDispatch[] = $meta;
        }

        $dispatchResult = $this->dispatchTicketMetaList($toDispatch, 'auto');

        SplitTicketSyncLogger::log('cron', 'auto sync dispatched', [
            'queuedCount'         => count($dispatchResult['queued']),
            'skippedCount'        => count($skipped) + count($dispatchResult['skipped']),
            'failedCount'         => count($dispatchResult['failed']),
            'stoppedByNodeBusy'   => $dispatchResult['stoppedByNodeBusy'],
            'nodeQueue'           => SplitNodeHealthService::getQueueSnapshot(),
        ]);

        return [
            'queued'            => $dispatchResult['queued'],
            'skipped'           => array_merge($skipped, $dispatchResult['skipped']),
            'failed'            => $dispatchResult['failed'],
            'stoppedByNodeBusy' => $dispatchResult['stoppedByNodeBusy'],
        ];
    }

    /**
     * 查询仍在同步中的工单 ID（依据 runtime 文件锁）
     *
     * @param int[] $ticketIds
     * @return int[]
     */
    public function filterSyncingIds(array $ticketIds): array
    {
        $lockService = new SplitTicketSyncLockService();
        $syncing = [];
        foreach ($ticketIds as $rawId) {
            $ticketId = (int) $rawId;
            if ($ticketId > 0 && $lockService->isLocked($ticketId)) {
                $syncing[] = $ticketId;
            }
        }

        return $syncing;
    }

    /**
     * @param list<array{id:int,ticket_type:string,ticket_url:string}> $metaList
     * @return array{
     *   queued:list<array<string,mixed>>,
     *   skipped:list<array<string,mixed>>,
     *   failed:list<array<string,mixed>>,
     *   stoppedByNodeBusy:bool
     * }
     */
    private function dispatchTicketMetaList(array $metaList, string $source): array
    {
        $php = $this->resolvePhpBinary();
        $think = $this->resolveThinkScript();
        $root = $this->resolveRootPath();

        /** @var list<array<string,mixed>> $queued */
        $queued = [];
        /** @var list<array<string,mixed>> $skipped */
        $skipped = [];
        /** @var list<array<string,mixed>> $failed */
        $failed = [];
        $stoppedByNodeBusy = false;

        /** @var list<array{id:int,ticket_type:string,ticket_url:string}> $apiFirstList */
        $apiFirstList = [];
        /** @var list<array{id:int,ticket_type:string,ticket_url:string}> $directList */
        $directList = [];
        /** @var array<string, list<array{id:int,ticket_type:string,ticket_url:string}>> $nodeGroups */
        $nodeGroups = [];

        foreach ($metaList as $meta) {
            $ticketType = (string) $meta['ticket_type'];
            if (SplitScrmSpiderFactory::isApiFirstSyncType($ticketType)) {
                $apiFirstList[] = $meta;
                continue;
            }
            if (!SplitScrmSpiderFactory::requiresNode($ticketType)) {
                $directList[] = $meta;
                continue;
            }
            $sessionKey = AntiBotConfigBuilder::resolveSessionKey(
                $ticketType,
                (string) $meta['ticket_url']
            );
            $nodeGroups[$sessionKey][] = $meta;
        }

        foreach ($apiFirstList as $meta) {
            $this->spawnSingleMeta($php, $think, $root, $meta, $source, $queued, $failed);
        }

        foreach ($directList as $meta) {
            $this->spawnSingleMeta($php, $think, $root, $meta, $source, $queued, $failed);
        }

        foreach ($nodeGroups as $sessionKey => $group) {
            if ($group === []) {
                continue;
            }

            /** @var list<array{id:int,ticket_type:string,ticket_url:string}> $nodeReady */
            $nodeReady = [];
            foreach ($group as $meta) {
                $ticketType = (string) $meta['ticket_type'];
                if (!SplitNodeHealthService::canAcceptWorkForTicketType($ticketType)) {
                    $stoppedByNodeBusy = true;
                    $skipped[] = $this->buildDispatchEntry(
                        (int) $meta['id'],
                        $ticketType,
                        (string) $meta['ticket_url'],
                        'Node 队列繁忙，本轮未投递'
                    );
                    continue;
                }
                $nodeReady[] = $meta;
            }

            if ($nodeReady === []) {
                continue;
            }

            if (count($nodeReady) === 1) {
                $this->spawnSingleMeta($php, $think, $root, $nodeReady[0], $source, $queued, $failed);
                continue;
            }

            $ids = array_map(static function (array $row): int {
                return (int) $row['id'];
            }, $nodeReady);

            if ($this->spawnCliSequential($php, $think, $root, $ids, $source)) {
                foreach ($nodeReady as $meta) {
                    $queued[] = $this->buildQueuedEntry($meta, $sessionKey, 'node-group-serial');
                }
            } else {
                foreach ($nodeReady as $meta) {
                    $failed[] = $this->buildDispatchEntry(
                        (int) $meta['id'],
                        (string) $meta['ticket_type'],
                        (string) $meta['ticket_url'],
                        'CLI 投递失败'
                    );
                }
            }
        }

        return [
            'queued'            => $queued,
            'skipped'           => $skipped,
            'failed'            => $failed,
            'stoppedByNodeBusy' => $stoppedByNodeBusy,
        ];
    }

    /**
     * @param list<array<string,mixed>> $queued
     * @param list<array<string,mixed>> $failed
     * @param array{id:int,ticket_type:string,ticket_url:string} $meta
     */
    private function spawnSingleMeta(
        string $php,
        string $think,
        string $root,
        array $meta,
        string $source,
        array &$queued,
        array &$failed
    ): void {
        $ticketId = (int) $meta['id'];
        if ($this->spawnCli($php, $think, $root, $ticketId, $source)) {
            $queued[] = $this->buildQueuedEntry($meta, '', 'parallel');
            return;
        }

        $failed[] = $this->buildDispatchEntry(
            $ticketId,
            (string) $meta['ticket_type'],
            (string) $meta['ticket_url'],
            'CLI 投递失败'
        );
    }

    /**
     * @param array{id:int,ticket_type:string,ticket_url:string} $meta
     * @return array<string,mixed>
     */
    private function buildQueuedEntry(array $meta, string $sessionKey, string $mode): array
    {
        return [
            'ticketId'   => (int) $meta['id'],
            'ticketType' => (string) $meta['ticket_type'],
            'ticketUrl'  => (string) $meta['ticket_url'],
            'sessionKey' => $sessionKey,
            'mode'       => $mode,
            'dispatched' => true,
        ];
    }

    /**
     * @return array{ticketId:int,ticketType:string,ticketUrl:string,reason:string}
     */
    private function buildDispatchEntry(int $ticketId, string $ticketType, string $ticketUrl, string $reason): array
    {
        return [
            'ticketId'   => $ticketId,
            'ticketType' => $ticketType,
            'ticketUrl'  => $ticketUrl,
            'reason'     => $reason,
        ];
    }

    /**
     * @param int[] $ticketIds
     * @return list<array{id:int,ticket_type:string,ticket_url:string}>
     */
    private function loadTicketMetaList(array $ticketIds): array
    {
        $ids = [];
        foreach ($ticketIds as $rawId) {
            $ticketId = (int) $rawId;
            if ($ticketId > 0) {
                $ids[] = $ticketId;
            }
        }
        if ($ids === []) {
            return [];
        }

        $rows = Db::name('split_ticket')
            ->where('id', 'in', $ids)
            ->field('id,ticket_type,ticket_url,manual_manage')
            ->select();

        $list = [];
        foreach ($rows as $row) {
            $list[] = [
                'id'             => (int) $row['id'],
                'ticket_type'    => (string) ($row['ticket_type'] ?? ''),
                'ticket_url'     => (string) ($row['ticket_url'] ?? ''),
                'manual_manage'  => (int) ($row['manual_manage'] ?? 0),
            ];
        }

        return $list;
    }

    /**
     * @param array<string,mixed>|object $ticket
     * @return array{id:int,ticket_type:string,ticket_url:string}|null
     */
    private function normalizeTicketMeta($ticket): ?array
    {
        if (is_object($ticket) && method_exists($ticket, 'toArray')) {
            $ticket = $ticket->toArray();
        }
        if (!is_array($ticket)) {
            return null;
        }
        $ticketId = (int) ($ticket['id'] ?? 0);
        if ($ticketId <= 0) {
            return null;
        }

        return [
            'id'          => $ticketId,
            'ticket_type' => (string) ($ticket['ticket_type'] ?? ''),
            'ticket_url'  => (string) ($ticket['ticket_url'] ?? ''),
        ];
    }

    /**
     * 后台启动 split:sync-tickets CLI（单工单，独立进程）
     */
    private function spawnCli(string $php, string $think, string $root, int $ticketId, string $source): bool
    {
        $logDir = $this->resolveLogDir();
        $logFile = $logDir . 'cli_' . $ticketId . '_' . date('YmdHis') . '.log';

        $syncMode = $this->normalizeSyncModeForCli($source);
        $inner = sprintf(
            '%s %s split:sync-tickets --ticket=%d --mode=%s >> %s 2>&1',
            escapeshellarg($php),
            escapeshellarg($think),
            $ticketId,
            $syncMode,
            escapeshellarg($logFile)
        );

        return $this->runBackgroundCommand($root, $inner, $ticketId, $source . '-parallel');
    }

    /**
     * 同一 sessionKey 内串行执行多个 CLI（单 nohup 子 shell）
     *
     * @param int[] $ticketIds
     */
    private function spawnCliSequential(string $php, string $think, string $root, array $ticketIds, string $source): bool
    {
        if ($ticketIds === []) {
            return false;
        }

        $logDir = $this->resolveLogDir();
        $parts = [];
        foreach ($ticketIds as $ticketId) {
            $ticketId = (int) $ticketId;
            if ($ticketId <= 0) {
                continue;
            }
            $logFile = $logDir . 'cli_' . $ticketId . '_' . date('YmdHis') . '.log';
            $syncMode = $this->normalizeSyncModeForCli($source);
            $parts[] = sprintf(
                '%s %s split:sync-tickets --ticket=%d --mode=%s >> %s 2>&1',
                escapeshellarg($php),
                escapeshellarg($think),
                $ticketId,
                $syncMode,
                escapeshellarg($logFile)
            );
        }
        if ($parts === []) {
            return false;
        }

        $inner = implode('; ', $parts);

        return $this->runBackgroundCommand($root, $inner, $ticketIds, $source . '-node-serial');
    }

    /**
     * @param int|int[] $ticketRef 日志用 ticketId 或 ID 列表
     */
    private function runBackgroundCommand(string $root, string $inner, $ticketRef, string $mode): bool
    {
        if ($this->isWindows()) {
            $command = sprintf('start /B cmd /C %s', $inner);
        } else {
            $command = sprintf('cd %s && nohup bash -c %s &', escapeshellarg($root), escapeshellarg($inner));
        }

        if ($this->canUseExec()) {
            @exec($command, $output, $exitCode);
            SplitTicketSyncLogger::log('dispatch', 'cli spawn exec', [
                'ticketRef' => $ticketRef,
                'mode'      => $mode,
                'command'   => $command,
                'exitCode'  => $exitCode,
            ]);
            return true;
        }

        if ($this->canUseShellExec()) {
            @shell_exec($command);
            SplitTicketSyncLogger::log('dispatch', 'cli spawn shell_exec', [
                'ticketRef' => $ticketRef,
                'mode'      => $mode,
                'command'   => $command,
            ]);
            return true;
        }

        SplitTicketSyncLogger::log('dispatch', 'cli spawn failed: exec disabled', [
            'ticketRef' => $ticketRef,
            'mode'      => $mode,
        ]);

        return false;
    }

    private function resolvePhpBinary(): string
    {
        if (method_exists(SplitSyncConfigService::class, 'getCliPhpBinary')) {
            return SplitSyncConfigService::getCliPhpBinary();
        }

        return $this->resolvePhpBinaryFallback();
    }

    /**
     * 当 SplitSyncConfigService 未部署 getCliPhpBinary 时的兜底解析
     */
    private function resolvePhpBinaryFallback(): string
    {
        if (defined('PHP_BINDIR') && PHP_BINDIR !== '') {
            $candidate = rtrim(PHP_BINDIR, '/\\') . DIRECTORY_SEPARATOR . 'php';
            if ($this->isUsableCliPhp($candidate)) {
                return $candidate;
            }
        }
        foreach (['/usr/bin/php', '/usr/local/bin/php'] as $candidate) {
            if ($this->isUsableCliPhp($candidate)) {
                return $candidate;
            }
        }
        return 'php';
    }

    private function isUsableCliPhp(string $path): bool
    {
        if (stripos($path, 'fpm') !== false) {
            return false;
        }
        if ($path === 'php') {
            return true;
        }
        return is_file($path) && is_executable($path);
    }

    private function resolveThinkScript(): string
    {
        $root = $this->resolveRootPath();
        return rtrim($root, '/\\') . DIRECTORY_SEPARATOR . 'think';
    }

    private function resolveRootPath(): string
    {
        if (defined('ROOT_PATH') && ROOT_PATH !== '') {
            return rtrim(ROOT_PATH, '/\\') . DIRECTORY_SEPARATOR;
        }

        return rtrim(dirname(__DIR__, 3), '/\\') . DIRECTORY_SEPARATOR;
    }

    private function resolveLogDir(): string
    {
        $runtime = defined('RUNTIME_PATH') ? RUNTIME_PATH : ($this->resolveRootPath() . 'runtime' . DIRECTORY_SEPARATOR);
        $dir = rtrim($runtime, '/\\') . DIRECTORY_SEPARATOR . 'split_ticket_sync' . DIRECTORY_SEPARATOR;
        if (!is_dir($dir)) {
            @mkdir($dir, 0755, true);
        }

        return $dir;
    }

    private function isWindows(): bool
    {
        return strtoupper(substr(PHP_OS, 0, 3)) === 'WIN';
    }

    /**
     * @return string[]
     */
    private function disabledFunctions(): array
    {
        $raw = (string) ini_get('disable_functions');
        if ($raw === '') {
            return [];
        }

        return array_filter(array_map('trim', explode(',', $raw)));
    }

    private function canUseExec(): bool
    {
        return function_exists('exec') && !in_array('exec', $this->disabledFunctions(), true);
    }

    private function canUseShellExec(): bool
    {
        return function_exists('shell_exec') && !in_array('shell_exec', $this->disabledFunctions(), true);
    }

    /**
     * 投递 CLI 时携带 --mode，供 sync_success_mode 区分自动/手动
     */
    private function normalizeSyncModeForCli(string $source): string
    {
        return in_array($source, ['auto', 'manual'], true) ? $source : 'manual';
    }
}