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

<?php

declare(strict_types=1);

namespace app\common\service;

use think\Db;

/**
 * 工单手工同步 CLI 后台投递
 *
 * Web 请求仅负责鉴权与投递，实际 Spider 在独立 CLI 进程中执行，
 * 避免长时间占用 PHP-FPM 与 Session 锁导致后台其它页面无法打开。
 *
 * Node 类型（A2C / Whatshub 等）在同一 nohup 子 shell 内串行执行，避免并行 Chrome 压垮温池。
 */
class SplitTicketSyncDispatchService
{
    /**
     * 投递手工同步任务到后台 CLI
     *
     * @param int[] $ticketIds 已通过权限校验的工单 ID
     * @return array{queued:int[],skipped:int[],failed:int[]}
     */
    public function dispatchManual(array $ticketIds): array
    {
        $queued = [];
        $skipped = [];
        $failed = [];

        $php = $this->resolvePhpBinary();
        $think = $this->resolveThinkScript();
        $root = $this->resolveRootPath();
        $lockService = new SplitTicketSyncLockService();
        $nodeTypeMap = array_flip(SplitScrmSpiderFactory::listNodeSyncTypes());
        $typeMap = $this->loadTicketTypeMap($ticketIds);

        /** @var int[] $nodeTicketIds */
        $nodeTicketIds = [];
        /** @var int[] $directTicketIds */
        $directTicketIds = [];

        foreach ($ticketIds as $rawId) {
            $ticketId = (int) $rawId;
            if ($ticketId <= 0) {
                continue;
            }

            if ($lockService->isLocked($ticketId)) {
                $skipped[] = $ticketId;
                continue;
            }

            $ticketType = (string) ($typeMap[$ticketId] ?? '');
            if ($ticketType !== '' && isset($nodeTypeMap[$ticketType])) {
                $nodeTicketIds[] = $ticketId;
            } else {
                $directTicketIds[] = $ticketId;
            }
        }

        foreach ($directTicketIds as $ticketId) {
            if ($this->spawnCli($php, $think, $root, $ticketId)) {
                $queued[] = $ticketId;
            } else {
                $failed[] = $ticketId;
            }
        }

        if ($nodeTicketIds !== []) {
            $spawned = $this->spawnCliSequential($php, $think, $root, $nodeTicketIds);
            if ($spawned) {
                foreach ($nodeTicketIds as $ticketId) {
                    $queued[] = $ticketId;
                }
            } else {
                foreach ($nodeTicketIds as $ticketId) {
                    $failed[] = $ticketId;
                }
            }
        }

        SplitTicketSyncLogger::log('web', 'manual sync dispatched', [
            'queued'       => $queued,
            'skipped'      => $skipped,
            'failed'       => $failed,
            'nodeSerial'   => $nodeTicketIds,
            'directParallel' => $directTicketIds,
            'phpCli'       => $php,
        ]);

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

    /**
     * 查询仍在同步中的工单 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 int[] $ticketIds
     * @return array<int, string> ticketId => ticket_type
     */
    private function loadTicketTypeMap(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')
            ->select();

        $map = [];
        foreach ($rows as $row) {
            $map[(int) $row['id']] = (string) ($row['ticket_type'] ?? '');
        }

        return $map;
    }

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

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

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

    /**
     * Node 类型工单：单条 nohup 内串行执行多个 CLI，避免并行 Chrome
     *
     * @param int[] $ticketIds
     */
    private function spawnCliSequential(string $php, string $think, string $root, array $ticketIds): 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';
            $parts[] = sprintf(
                '%s %s split:sync-tickets --ticket=%d >> %s 2>&1',
                escapeshellarg($php),
                escapeshellarg($think),
                $ticketId,
                escapeshellarg($logFile)
            );
        }
        if ($parts === []) {
            return false;
        }

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

        return $this->runBackgroundCommand($root, $inner, $ticketIds, '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('web', 'cli spawn exec', [
                'ticketRef' => $ticketRef,
                'mode'      => $mode,
                'command'   => $command,
                'exitCode'  => $exitCode,
            ]);
            return true;
        }

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

        SplitTicketSyncLogger::log('web', '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);
    }
}