<?php

declare(strict_types=1);

namespace base\tools\controller\timer;

use app\admin\model\XhprofRun;
use app\admin\model\XhprofRunDetail;
use app\admin\model\XhprofRunWatch;
use app\admin\model\XhprofWatch;
use app\common\controller\TimerController;
use think\facade\Db;
use think\facade\Log;

/**
 * XHProf jsonl 导入定时任务（Base 层）.
 *
 * 流程：
 *   1. 扫描 XHPROF_FILE_DIR 下 runs-*.jsonl 按文件名序（旧日期先导）
 *   2. 依赖倒置：实例化 app 层 Parser / Reader；watch 正则 per 批次编译一次
 *   3. 每文件：增量读取（position 按 node_id+file_path 定位，inode 变化归零）→
 *      run 逐条 insert 拿自增 id → detail chunk 20 insertAll →
 *      run_watch 物化聚合 insertAll；同一事务内 insertBatch + savePosition（消崩溃重复窗口）
 *   4. 文件导至 EOF 且 position 落盘后 unlink（只删已导完文件）
 *
 * 门控约定：本任务不受 XHPROF_ENABLE 门控——开关只管采集端，
 * 已落盘的残留文件必须能导完（关闭采集后存量数据照常入库）。
 *
 * 目录单一来源：getenv('XHPROF_FILE_DIR')，与采集端 bootstrap 读同一变量，
 * 禁止与 ThinkPHP env() 混用（两处漂移会导致导入读不到文件）。
 *
 * 业务侧如需定制导入逻辑，重写 app/tools/controller/timer/XhprofLogImport.php 对应方法即可拦截。
 */
class XhprofLogImportBase extends TimerController
{
    /**
     * 防刷间隔（秒）.
     * 仅做控制器侧防刷，不影响定时器侧 frequency 节流。
     */
    protected $frequency = 60;

    /**
     * 并发上限 cap：单分片（id=0）。jsonl 是节点本地文件，多节点各自导自己的。
     */
    protected $concurrency = 1;

    /**
     * run 行缓冲上限（同时是 detail insertAll 的 chunk；单行实测 ~100KB，20 条一批防单条 SQL 过大）.
     */
    protected const RUN_BATCH = 20;

    /**
     * 单文件单轮最大读取行数（行大，2000 行 ≈ 200MB 上界已足够单轮消化；未读完下轮继续）.
     */
    protected const READ_MAX_LINES = 2000;

    /**
     * parse_fail_samples 最大记录条数.
     */
    protected const FAIL_SAMPLES_MAX = 10;

    public function do()
    {
        // 注意：不受 XHPROF_ENABLE 门控（开关只管采集，残留文件必须能导完）
        $dir = $this->fileDir();
        $files = glob($dir . '/runs-*.jsonl');
        if ($files === false || $files === []) {
            return json_encode([
                'files' => 0, 'runs' => 0, 'watches' => 0, 'fails' => 0, 'deleted' => 0,
            ], JSON_UNESCAPED_UNICODE);
        }
        sort($files); // 按文件名序（runs-{Ymd} 字典序即日期序），旧日期先导

        // 依赖倒置：使用 app 层入口类（业务侧可重写拦截）
        $parser = new \app\common\service\XhprofProfileParserService();
        $reader = new \app\common\service\XhprofLogReaderService();

        // watch 正则 per 批次编译：本节点启用的规则一次编译，导入全程复用
        $watchRows = XhprofWatch::field('id,regex')->where('status', 1)->select()->toArray();
        $compiled = $parser->compileWatchRegexes($watchRows);
        $patterns = $compiled['patterns'];

        $totalFiles = 0;
        $totalRuns = 0;
        $totalWatches = 0;
        $totalFails = $compiled['fails'];
        $deleted = 0;
        $errors = [];

        foreach ($files as $file) {
            try {
                $stats = $this->processFile($parser, $reader, $file, $patterns);
                $totalFiles++;
                $totalRuns += $stats['runs'];
                $totalWatches += $stats['watches'];
                $totalFails += $stats['fails'];
                if ($stats['deleted']) {
                    $deleted++;
                }
            } catch (\Throwable $e) {
                Log::error('xhprof_log_import: 处理文件异常 ' . $file . ' - ' . $e->getMessage());
                $errors[] = 'error:' . basename($file) . ':' . mb_substr($e->getMessage(), 0, 200);
            }
        }

        return json_encode([
            'files' => $totalFiles,
            'runs' => $totalRuns,
            'watches' => $totalWatches,
            'fails' => $totalFails,
            'deleted' => $deleted,
            'errors' => $errors,
        ], JSON_UNESCAPED_UNICODE);
    }

    /**
     * jsonl 落盘目录（与采集端 bootstrap getenv 读同一变量，禁止 ThinkPHP env()）.
     */
    protected function fileDir(): string
    {
        return getenv('XHPROF_FILE_DIR') ?: '/var/www/html/runtime/xhprof';
    }

    /**
     * 处理单个文件：增量读取、解析、批量入库（事务内 savePosition）、导完删除.
     *
     * @return array{runs: int, watches: int, fails: int, deleted: bool}
     */
    protected function processFile(
        \app\common\service\XhprofProfileParserService $parser,
        \app\common\service\XhprofLogReaderService $reader,
        string $file,
        array $patterns
    ): array {
        $runs = 0;
        $watches = 0;
        $fails = 0;
        $samples = [];
        $buffer = [];

        // Reader 是 Generator，让其自然走完（maxLines / EOF 自然结束，进度由本类事务内落盘）
        foreach ($reader->read($file, self::READ_MAX_LINES) as $line) {
            $parsed = $parser->parse($line);
            if ($parsed === null) {
                $fails++;
                $this->addFailSample($samples, $line);
                continue;
            }
            if ($parsed['degraded']) {
                // main() 缺失：行仍入库（降级 0），但计 fail 留痕
                $fails++;
                $this->addFailSample($samples, $line);
            }

            $buffer[] = [
                'run' => $this->buildRunRow($parsed),
                'raw' => $line,
                'profile' => $parsed['profile'],
            ];

            if (count($buffer) >= self::RUN_BATCH) {
                $flushed = $this->flushBatch($parser, $reader, $file, $buffer, $patterns, $fails, $samples);
                $runs += $flushed['runs'];
                $watches += $flushed['watches'];
                $buffer = [];
            }
        }

        // flush 剩余；buffer 为空时也补一次 position 落盘（推进被失败行消费掉的区间）
        if (!empty($buffer)) {
            $flushed = $this->flushBatch($parser, $reader, $file, $buffer, $patterns, $fails, $samples);
            $runs += $flushed['runs'];
            $watches += $flushed['watches'];
        } else {
            $position = $reader->getCurrentPosition();
            $reader->savePosition(
                $file,
                $position['inode'],
                $position['offset'],
                $position['last_line_hash'],
                $fails,
                $this->encodeSamples($samples)
            );
        }

        // 超长行丢弃并入 fails（Reader 整行丢弃计数）
        $fails += $reader->getDroppedLines();

        // 文件导至 EOF 且 position 落盘后才可删除（只删已导完文件，未导完/尾部半行不删）
        $isDeleted = false;
        if ($reader->reachedEof()) {
            clearstatcache(true, $file);
            $position = $reader->getCurrentPosition();
            $size = filesize($file);
            if ($size !== false && $position['offset'] >= $size) {
                $isDeleted = @unlink($file);
                if ($isDeleted) {
                    Log::info('xhprof_log_import: 文件已导完删除 ' . $file);
                }
            }
        }

        Log::info("xhprof_log_import: 文件 {$file} 完成 runs={$runs} watches={$watches} fails={$fails}");

        return ['runs' => $runs, 'watches' => $watches, 'fails' => $fails, 'deleted' => $isDeleted];
    }

    /**
     * 构造 run 入库行（字段对齐 ul_xhprof_run，单位 μs/字节）.
     */
    protected function buildRunRow(array $parsed): array
    {
        return [
            'node_id' => \app\common\service\HostService::getNodeId(),
            'url' => $parsed['url'],
            'simple_url' => $parsed['simple_url'],
            'method' => $parsed['method'],
            'wall_time' => $parsed['wall_time'],
            'cpu_time' => $parsed['cpu_time'],
            'memory_peak' => $parsed['memory_peak'],
            'request_time' => $parsed['request_time'],
            'create_time' => time(),
        ];
    }

    /**
     * 批量入库：同一事务内 run 逐条 insert（拿自增 id）→ detail/run_watch insertAll → savePosition.
     *
     * 事务语义：入库与 position 落盘要么同时生效、要么同时回滚——
     * 崩溃恢复后从 position 处重读，不会重复导入已提交批次。
     *
     * @param array $buffer [['run' => array, 'raw' => string, 'profile' => array], ...]
     *
     * @return array{runs: int, watches: int}
     */
    protected function flushBatch(
        \app\common\service\XhprofProfileParserService $parser,
        \app\common\service\XhprofLogReaderService $reader,
        string $file,
        array $buffer,
        array $patterns,
        int $fails,
        array $samples
    ): array {
        // 捕获当前安全偏移（最后一条已 yield 完整行末尾），事务内与数据同落盘
        $position = $reader->getCurrentPosition();
        $samplesJson = $this->encodeSamples($samples);
        $runs = 0;
        $watches = 0;

        Db::transaction(function () use ($parser, $reader, $file, $buffer, $patterns, $position, $fails, $samplesJson, &$runs, &$watches) {
            $detailBatch = [];
            $watchBatch = [];
            $now = time();

            foreach ($buffer as $item) {
                $runId = (int) XhprofRun::insertGetId($item['run']);
                $runs++;

                $detailBatch[] = [
                    'id' => $runId,
                    'profile_data' => $item['raw'],
                ];

                foreach ($parser->materializeWatches($item['profile'], $patterns) as $watchId => $agg) {
                    $watchBatch[] = [
                        'run_id' => $runId,
                        'watch_id' => $watchId,
                        'ct' => $agg['ct'],
                        'wt_sum' => $agg['wt_sum'],
                        'wt_max' => $agg['wt_max'],
                        'create_time' => $now,
                    ];
                    $watches++;
                }
            }

            if (!empty($detailBatch)) {
                XhprofRunDetail::insertAll($detailBatch);
            }
            if (!empty($watchBatch)) {
                XhprofRunWatch::insertAll($watchBatch);
            }

            $reader->savePosition(
                $file,
                $position['inode'],
                $position['offset'],
                $position['last_line_hash'],
                $fails,
                $samplesJson
            );
        });

        return ['runs' => $runs, 'watches' => $watches];
    }

    /**
     * 收集解析失败样本（截断至 500 字符，最多 FAIL_SAMPLES_MAX 条）.
     */
    protected function addFailSample(array &$samples, string $line): void
    {
        if (count($samples) >= self::FAIL_SAMPLES_MAX) {
            return;
        }
        $samples[] = mb_substr($line, 0, 500);
    }

    /**
     * 失败样本编码为 json（空样本返回 null，与 position 列 nullable 对齐）.
     */
    protected function encodeSamples(array $samples): ?string
    {
        return empty($samples) ? null : json_encode($samples, JSON_UNESCAPED_UNICODE);
    }
}