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

<?php

namespace app\common\service\ai;

use app\common\model\ai\AiKbDocument;
use app\common\model\ai\AiKbSyncJob;
use think\facade\Db;

class KbSyncService
{
    public static function enqueue(
        string $domain,
        string $subtype,
        int $sourceId,
        string $action = AiKbSyncJob::ACTION_UPSERT,
        int $priority = 100,
        ?int $scheduledAt = null
    ): void {
        if ($domain === '' || $subtype === '' || $sourceId <= 0) {
            return;
        }

        $scheduledAt = $scheduledAt ?: time();
        $pending = AiKbSyncJob::where([
            'domain' => $domain,
            'subtype' => $subtype,
            'source_id' => $sourceId,
        ])->whereIn('status', [AiKbSyncJob::STATUS_PENDING, AiKbSyncJob::STATUS_PROCESSING])
            ->order('id', 'desc')
            ->findOrEmpty();

        if (!$pending->isEmpty() && (int) $pending->status === AiKbSyncJob::STATUS_PENDING) {
            $pending->action = $action;
            $pending->priority = $priority;
            $pending->scheduled_at = $scheduledAt;
            $pending->error_msg = '';
            $pending->update_time = time();
            $pending->save();
            return;
        }

        AiKbSyncJob::create([
            'domain' => $domain,
            'subtype' => $subtype,
            'source_id' => $sourceId,
            'action' => $action,
            'priority' => $priority,
            'status' => AiKbSyncJob::STATUS_PENDING,
            'retry_count' => 0,
            'error_msg' => '',
            'scheduled_at' => $scheduledAt,
            'processed_at' => 0,
            'create_time' => time(),
            'update_time' => time(),
        ]);
    }

    public static function markArticleBackfillJobs(int $limit = 200): void
    {
        self::enqueueUnsyncedRows(
            'article',
            'article',
            'article',
            $limit,
            85,
            'COALESCE(s.update_time, s.create_time, 0)'
        );
    }

    public static function markLotteryBackfillJobs(int $limit = 200): void
    {
        self::enqueueMissingLotteryRows('draw_result', 'lottery_draw_result', $limit, 80);
        self::enqueueMissingLotteryRows('ai_history', 'lottery_ai_analysis', $limit, 90);
    }

    public static function markMatchBackfillJobs(int $limit = 200): void
    {
        self::enqueueUnsyncedRows(
            'match',
            'event',
            'match',
            $limit,
            60,
            'COALESCE(s.update_time, s.create_time, s.match_time, 0)'
        );
    }

    private static function enqueueMissingLotteryRows(string $subtype, string $table, int $limit, int $priority): void
    {
        self::enqueueUnsyncedRows(
            'lottery',
            $subtype,
            $table,
            $limit,
            $priority,
            'COALESCE(s.update_time, s.create_time, 0)'
        );
    }

    private static function enqueueUnsyncedRows(
        string $domain,
        string $subtype,
        string $table,
        int $limit,
        int $priority,
        string $sourceUpdatedExpr
    ): void {
        try {
            $query = Db::name($table)->alias('s')
                ->leftJoin('ai_kb_document d', "d.domain = '{$domain}' AND d.subtype = '{$subtype}' AND d.source_id = s.id")
                ->where('s.id', '>', 0)
                ->whereRaw("(d.id IS NULL OR d.is_active <> 1 OR d.source_updated_at < {$sourceUpdatedExpr})");

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

            $rows = $query
                ->order('s.id', 'desc')
                ->limit($limit)
                ->column('s.id');

            foreach ($rows as $sourceId) {
                self::enqueue($domain, $subtype, (int) $sourceId, AiKbSyncJob::ACTION_UPSERT, $priority);
            }
        } catch (\Throwable $e) {
            // Ignore in environments where KB tables are not ready yet.
        }
    }

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

        if ($domain === 'match') {
            $query->where('s.is_show', 1);
        }
    }

    public static function markDocumentInactive(string $domain, string $subtype, int $sourceId): void
    {
        AiKbDocument::where([
            'domain' => $domain,
            'subtype' => $subtype,
            'source_id' => $sourceId,
        ])->update([
            'is_active' => 0,
            'update_time' => time(),
        ]);
    }
}