$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(), ]); } }