<?php
namespace app\command;

use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Log;
use think\Db;
use think\Queue;

/**
 * 重试失败的埋点数据 PV UV 等指标
 * 每1个小时执行一次
 *
 * 任务描述：每1个小时执行一次，重试失败的埋点数据 PV UV 等指标
*/
class TrackFailedRetry extends Command
{
    protected $output = null;

    // 配置任务名称、描述（自定义，便于后续调用）
    protected function configure()
    {
        $this->setName('shopro:track-failed-retry')->setDescription('每1个小时执行一次');
    }

    /**
     * 自定义日志方法
     * @param string $level 日志级别: info/warning/error
     * @param string $message 日志内容
     */
    protected function customLog($level, $message) {
        $this->output->writeln($message);

        custom_log($message, 'stat_track_failed_retry', $level);
    }

    protected function execute(Input $input, Output $output) {
        $this->output = $output;

        $this->customLog('info', '========== [' . date('Y-m-d H:i:s') . '] 开始执行重试失败埋埋点数据任务 ==========');

        $this->retryFailed();

        $this->customLog('info', '========== [' . date('Y-m-d H:i:s') . '] 重试失败埋埋点数据任务执行完成========== ');

    }


    /**
     * 分页批量重试失败数据
     */
    public function retryFailed()
    {
        // 每次处理的数量
        $batchSize = 500;
        $page = 1;
        $totalProcessed = 0;
        $totalSuccess = 0;
        $totalFailed = 0;

        while (true) {
            // 分页查询失败数据
            $failedList = Db::name('shopro_miniprogram_track_failed')
                ->where('status', 0)
                ->page($page, $batchSize)
                ->select();

            if (empty($failedList)) {
                break;
            }

            foreach ($failedList as $item) {
                try {
                    $data = json_decode($item['data'], true);

                    // 验证数据格式
                    if (empty($data) || !isset($data['data'])) {
                        throw new \Exception('数据格式错误');
                    }

                    $retryPayload = $this->filterRetryPayload($data);

                    if (empty($retryPayload['data'])) {
                        Db::name('shopro_miniprogram_track_failed')
                            ->where('id', $item['id'])
                            ->update([
                                'status' => 1,
                                'error' => '同一操作已入库，跳过重试',
                                'update_time' => time()
                            ]);

                        $totalSuccess++;
                        $totalProcessed++;
                        continue;
                    }

                    // 重新推入队列
                    Queue::push('\addons\shopro\job\TrackProcessJob@fire', $retryPayload, 'shopro-high');

                    // 更新状态为已处理
                    Db::name('shopro_miniprogram_track_failed')
                        ->where('id', $item['id'])
                        ->update(['status' => 1, 'update_time' => time()]);

                    $totalSuccess++;

                } catch (\Exception $e) {
                    // 更新状态为已放弃
                    Db::name('shopro_miniprogram_track_failed')
                        ->where('id', $item['id'])
                        ->update([
                            'status' => 2,
                            'error' => $e->getMessage(),
                            'update_time' => time(),
                            'attempts' => Db::raw('attempts + 1')
                        ]);

                    $totalFailed++;
                }

                $totalProcessed++;
            }

            // 输出进度信息
            $this->customLog('info',  "已处理: {$totalProcessed} | 成功: {$totalSuccess} | 失败: {$totalFailed}");;

            $page++;
        }

        $this->customLog('info',  "重试完成！总计处理: {$totalProcessed} | 成功: {$totalSuccess} | 失败: {$totalFailed}");
    }

    /**
     * 过滤已经入库的埋点，仅重试未入库数据
     */
    protected function filterRetryPayload($payload)
    {
        $trackList = isset($payload['data']) && is_array($payload['data']) ? $payload['data'] : [];
        $clientIp = $payload['ip'] ?? '';
        $jobTimestamp = isset($payload['timestamp']) ? (int)$payload['timestamp'] : time();
        $pendingList = [];
        $seen = [];

        foreach ($trackList as $trackItem) {
            $trackRow = $this->buildTrackRow($trackItem, $clientIp, $jobTimestamp);
            $dedupKey = $this->buildTrackDedupKey($trackRow);

            if (isset($seen[$dedupKey])) {
                continue;
            }
            $seen[$dedupKey] = true;

            if ($this->trackExists($trackRow)) {
                continue;
            }

            $pendingList[] = $trackItem;
        }

        $payload['data'] = $pendingList;
        return $payload;
    }

    /**
     * 构建埋点行数据
     */
    protected function buildTrackRow($trackItem, $clientIp, $jobTimestamp)
    {
        return [
            'user_id' => $trackItem['user_id'] ?? 0,
            'openid' => $trackItem['openid'] ?? '',
            'session_id' => $trackItem['session_id'] ?? '',
            'event_type' => $trackItem['event_type'] ?? '',
            'event_code' => $trackItem['event_code'] ?? '',
            'page_path' => $trackItem['page_path'] ?? '',
            'page_title' => $trackItem['page_title'] ?? '',
            'params' => !empty($trackItem['params']) ? json_encode($trackItem['params']) : '',
            'ip' => $clientIp,
            'create_time' => isset($trackItem['create_time']) ? (int)($trackItem['create_time'] / 1000) : $jobTimestamp
        ];
    }

    /**
     * 构建埋点判重键
     */
    protected function buildTrackDedupKey($trackRow)
    {
        return md5(implode('|', [
            (string)$trackRow['user_id'],
            (string)$trackRow['openid'],
            (string)$trackRow['session_id'],
            (string)$trackRow['event_type'],
            (string)$trackRow['event_code'],
            (string)$trackRow['page_path'],
            (string)$trackRow['page_title'],
            (string)$trackRow['params'],
            (string)$trackRow['ip'],
            (string)$trackRow['create_time'],
        ]));
    }

    /**
     * 判断埋点是否已存在
     */
    protected function trackExists($trackRow)
    {
        return Db::name('shopro_miniprogram_track')
            ->where('user_id', $trackRow['user_id'])
            ->where('openid', $trackRow['openid'])
            ->where('session_id', $trackRow['session_id'])
            ->where('event_type', $trackRow['event_type'])
            ->where('event_code', $trackRow['event_code'])
            ->where('page_path', $trackRow['page_path'])
            ->where('page_title', $trackRow['page_title'])
            ->where('params', $trackRow['params'])
            ->where('ip', $trackRow['ip'])
            ->where('create_time', $trackRow['create_time'])
            ->find() ? true : false;
    }


}
?>
