sbnews/server/app/common/service/ai/KbRedisQueueService.php

<?php

namespace app\common\service\ai;

use think\facade\Db;

class KbRedisQueueService
{
    public const STREAM_KEY = 'ai:kb:sync:stream';
    public const DEDUPE_PREFIX = 'ai:kb:sync:dedupe:';

    private const DEFAULT_DEDUPE_TTL = 1800;

    public static function enqueueRecent(int $limit = 1000, int $minutes = 30): array
    {
        $limit = max(1, $limit);
        $minutes = max(1, $minutes);
        $cutoff = time() - ($minutes * 60);
        $redis = self::redis();
        $summary = [
            'stream' => self::STREAM_KEY,
            'limit' => $limit,
            'minutes' => $minutes,
            'scanned' => 0,
            'queued' => 0,
            'deduped' => 0,
            'targets' => [],
        ];

        foreach (self::targets() as $target) {
            $result = self::enqueueTarget($redis, $target, $limit, $cutoff, $minutes * 60);
            $summary['targets'][$target['label']] = $result;
            $summary['scanned'] += $result['scanned'];
            $summary['queued'] += $result['queued'];
            $summary['deduped'] += $result['deduped'];
        }

        return $summary;
    }

    private static function enqueueTarget(\Redis $redis, array $target, int $limit, int $cutoff, int $dedupeTtl): array
    {
        $result = [
            'domain' => $target['domain'],
            'subtype' => $target['subtype'],
            'table' => $target['table'],
            'scanned' => 0,
            'queued' => 0,
            'deduped' => 0,
        ];

        $rows = self::recentUnsyncedRows($target, $limit, $cutoff);
        $result['scanned'] = count($rows);

        foreach ($rows as $row) {
            $sourceId = (int) ($row['source_id'] ?? 0);
            if ($sourceId <= 0) {
                continue;
            }

            $dedupeKey = self::dedupeKey($target['domain'], $target['subtype'], $sourceId);
            $setResult = $redis->rawCommand('SET', $dedupeKey, '1', 'EX', (string) max(60, $dedupeTtl), 'NX');
            if ($setResult !== true && strtoupper((string) $setResult) !== 'OK') {
                $result['deduped']++;
                continue;
            }

            $messageId = self::xAdd($redis, [
                'domain' => $target['domain'],
                'subtype' => $target['subtype'],
                'source_id' => (string) $sourceId,
                'source_updated_at' => (string) (int) ($row['source_updated_at'] ?? 0),
                'priority' => (string) (int) $target['priority'],
                'trace_id' => self::traceId($target['label'], $sourceId),
            ]);

            if ($messageId !== false && $messageId !== null && $messageId !== '') {
                $result['queued']++;
            }
        }

        return $result;
    }

    private static function recentUnsyncedRows(array $target, int $limit, int $cutoff): array
    {
        $sourceUpdatedExpr = $target['source_updated_expr'];
        $query = Db::name($target['table'])->alias('s')
            ->leftJoin(
                'ai_kb_document d',
                "d.domain = '{$target['domain']}' AND d.subtype = '{$target['subtype']}' AND d.source_id = s.id"
            )
            ->where('s.id', '>', 0)
            ->whereRaw("{$sourceUpdatedExpr} >= {$cutoff}")
            ->whereRaw("(d.id IS NULL OR d.is_active <> 1 OR d.source_updated_at < {$sourceUpdatedExpr})");

        self::applySourceFilters($query, $target);

        return $query
            ->fieldRaw("s.id AS source_id, {$sourceUpdatedExpr} AS source_updated_at")
            ->orderRaw("{$sourceUpdatedExpr} DESC, s.id DESC")
            ->limit($limit)
            ->select()
            ->toArray();
    }

    private static function applySourceFilters($query, array $target): void
    {
        if (($target['domain'] ?? '') === 'article') {
            $query->where('s.is_show', 1)->whereNull('s.delete_time');
        }

        if (($target['domain'] ?? '') === 'match') {
            $query->where('s.is_show', 1)->whereNull('s.delete_time');
        }

        if (($target['domain'] ?? '') === 'post') {
            $query->where('s.status', 1)->whereNull('s.delete_time');
        }

        foreach (($target['where'] ?? []) as $field => $value) {
            $query->where('s.' . $field, $value);
        }
    }

    private static function targets(): array
    {
        return [
            [
                'label' => 'crypto_article',
                'domain' => 'article',
                'subtype' => 'article',
                'table' => 'article',
                'priority' => 82,
                'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, 0)',
                'where' => ['cid' => 9],
            ],
            [
                'label' => 'article',
                'domain' => 'article',
                'subtype' => 'article',
                'table' => 'article',
                'priority' => 85,
                'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, 0)',
            ],
            [
                'label' => 'lottery_draw_result',
                'domain' => 'lottery',
                'subtype' => 'draw_result',
                'table' => 'lottery_draw_result',
                'priority' => 80,
                'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, 0)',
            ],
            [
                'label' => 'lottery_ai_history',
                'domain' => 'lottery',
                'subtype' => 'ai_history',
                'table' => 'lottery_ai_analysis',
                'priority' => 90,
                'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, 0)',
            ],
            [
                'label' => 'match',
                'domain' => 'match',
                'subtype' => 'event',
                'table' => 'match',
                'priority' => 60,
                'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, s.match_time, 0)',
            ],
            [
                'label' => 'post',
                'domain' => 'post',
                'subtype' => 'post',
                'table' => 'community_post',
                'priority' => 75,
                'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, 0)',
            ],
        ];
    }

    private static function redis(): \Redis
    {
        if (!class_exists(\Redis::class)) {
            throw new \RuntimeException('Redis 扩展不可用，无法写入 AI 知识库 Redis 队列');
        }

        $options = (array) config('cache.stores.redis');
        $redis = new \Redis();
        $host = (string) (env('AI_KB_REDIS_HOST', '') ?: ($options['host'] ?? '127.0.0.1'));
        $port = (int) (env('AI_KB_REDIS_PORT', '') ?: ($options['port'] ?? 6379));
        $timeout = (float) ($options['timeout'] ?? 0);
        $redis->connect($host, $port, $timeout);

        $password = (string) (env('AI_KB_REDIS_PASSWORD', '') ?: ($options['password'] ?? ''));
        if ($password !== '') {
            $redis->auth($password);
        }

        $select = (int) (env('AI_KB_REDIS_DB', '') ?: 4);
        $redis->select($select);

        return $redis;
    }

    private static function xAdd(\Redis $redis, array $fields)
    {
        if (method_exists($redis, 'xAdd')) {
            return $redis->xAdd(self::STREAM_KEY, '*', $fields);
        }

        $args = ['XADD', self::STREAM_KEY, '*'];
        foreach ($fields as $field => $value) {
            $args[] = (string) $field;
            $args[] = (string) $value;
        }

        return $redis->rawCommand(...$args);
    }

    private static function dedupeKey(string $domain, string $subtype, int $sourceId): string
    {
        return self::DEDUPE_PREFIX . $domain . ':' . $subtype . ':' . $sourceId;
    }

    private static function traceId(string $label, int $sourceId): string
    {
        try {
            $suffix = bin2hex(random_bytes(4));
        } catch (\Throwable $e) {
            $suffix = (string) mt_rand(100000, 999999);
        }

        return $label . '-' . $sourceId . '-' . date('YmdHis') . '-' . $suffix;
    }
}