Merge remote-tracking branch 'origin/master'

fs-live-app/src/main/java/com/fs/live/config/ProductionWordFilter.java

@@ -1,23 +1,41 @@
 package com.fs.live.config;
 
+import com.fs.common.enums.DataSourceType;
+import com.fs.common.utils.StringUtils;
+import com.fs.framework.datasource.DynamicDataSourceContextHolder;
+import com.fs.live.datasource.TenantDataSourceManager;
 import com.fs.live.service.ILiveSensitiveWordsService;
+import com.fs.tenant.domain.TenantInfo;
+import com.fs.tenant.service.TenantInfoService;
 import lombok.Getter;
-import org.apache.commons.lang3.StringUtils;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.Resource;
 import java.util.*;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
+@Slf4j
 @Component
 public class ProductionWordFilter implements InitializingBean {
 
     @Autowired
     private ILiveSensitiveWordsService liveSensitiveWordsService;
 
+    @Resource
+    private TenantDataSourceManager tenantDataSourceManager;
+    @Resource
+    private TenantInfoService tenantInfoService;
+
+    @Value("${saas.task.enabled}")
+    private boolean saasEnabled;
+
 
     private static class TrieNode {
         @Getter
@@ -77,15 +95,77 @@ public class ProductionWordFilter implements InitializingBean {
         executor.scheduleWithFixedDelay(this::reload, 1, 1, TimeUnit.HOURS);
     }
 
+    /**
+     * 重新加载敏感词。
+     * SaaS 模式下：遍历所有租户数据库，合并所有敏感词构建全局 Trie 树。
+     * 非 SaaS 模式下：从当前数据源加载。
+     */
     public synchronized void reload() {
+        if (saasEnabled) {
+            reloadForAllTenants();
+        } else {
+            reloadForCurrentDs();
+        }
+    }
+
+    /**
+     * SaaS 模式：遍历所有有效租户，合并各租户的敏感词。
+     */
+    private void reloadForAllTenants() {
+        List<TenantInfo> tenants = getValidTenants();
+        Set<String> allWords = new LinkedHashSet<>();
+        for (TenantInfo tenant : tenants) {
+            try {
+                tenantDataSourceManager.switchTenant(tenant);
+                List<String> words = liveSensitiveWordsService.selectAllWords();
+                if (words != null && !words.isEmpty()) {
+                    allWords.addAll(words);
+                }
+            } catch (Exception e) {
+                log.error("[SaaS] 加载租户 {} 敏感词失败", tenant.getId(), e);
+            } finally {
+                DynamicDataSourceContextHolder.clearDataSourceType();
+            }
+        }
+        TrieNode newRoot = new TrieNode();
+        allWords.forEach(word -> addWord(newRoot, word));
+        this.root = newRoot;
+        this.wordSources = new ArrayList<>(allWords);
+        log.info("[SaaS] 敏感词重载完成，共 {} 个租户，{} 个敏感词", tenants.size(), allWords.size());
+    }
+
+    /**
+     * 非 SaaS 模式：从当前数据源加载敏感词。
+     */
+    private void reloadForCurrentDs() {
         wordSources = liveSensitiveWordsService.selectAllWords();
         TrieNode newRoot = new TrieNode();
-        wordSources.stream()
-                .flatMap(source -> loadWords(source).stream())
-                .forEach(word -> addWord(newRoot, word));
+        if (wordSources != null) {
+            wordSources.stream()
+                    .flatMap(source -> loadWords(source).stream())
+                    .forEach(word -> addWord(newRoot, word));
+        }
         this.root = newRoot;
     }
 
+    private List<TenantInfo> getValidTenants() {
+        try {
+            DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.MASTER.name());
+            TenantInfo query = new TenantInfo();
+            query.setStatus(1);
+            List<TenantInfo> tenants = tenantInfoService.selectTenantInfoList(query);
+            if (tenants == null || tenants.isEmpty()) {
+                return Collections.emptyList();
+            }
+            Date now = new Date();
+            return tenants.stream()
+                    .filter(t -> t.getExpireTime() == null || !t.getExpireTime().before(now))
+                    .collect(Collectors.toList());
+        } finally {
+            DynamicDataSourceContextHolder.clearDataSourceType();
+        }
+    }
+
     public FilterResult filter(String text) {
         StringBuilder result = new StringBuilder();
         Set<String> foundWords = new HashSet<>();
@@ -132,8 +212,7 @@ public class ProductionWordFilter implements InitializingBean {
     }
 
     private List<String> loadWords(String source) {
-        // 示例：实际应从 source（如 URL 或文件路径）读取
-        return Collections.singletonList(source); // 替换为真实逻辑
+        return Collections.singletonList(source);
     }
 
 

fs-live-app/src/main/java/com/fs/live/datasource/TenantDataSourceManager.java

@@ -41,6 +41,25 @@ public class TenantDataSourceManager {
         DynamicDataSourceContextHolder.setDataSourceType(tenantKey);
     }
 
+    /**
+     * 判断指定数据源是否已创建。
+     */
+    public boolean hasDataSource(String dsKey) {
+        return TENANT_DS_CACHE.containsKey(dsKey);
+    }
+
+    /**
+     * 获取指定租户的数据源。
+     */
+    public DataSource getTenantDataSource(Long tenantId) {
+        String tenantKey = buildTenantKey(tenantId);
+        DataSource ds = TENANT_DS_CACHE.get(tenantKey);
+        if (ds == null) {
+            throw new IllegalStateException("租户数据源不存在: " + tenantKey);
+        }
+        return ds;
+    }
+
     /**
      * 清理当前线程的数据源标记。
      */

fs-live-app/src/main/java/com/fs/live/task/LiveCompletionPointsTask.java

@@ -35,6 +35,14 @@ public class LiveCompletionPointsTask {
     @Value("${saas.task.enabled}")
     private boolean saasTaskEnabled;
 
+    private void runWithSaaSTenant(String taskName, Runnable action) {
+        if (saasTaskEnabled && !TenantTaskRunner.isInTenantExecution()) {
+            tenantTaskRunner.runForEachTenant(taskName, action);
+            return;
+        }
+        action.run();
+    }
+
     /**
      * 定时检查观看时长并创建完课记录（兜底机制）
      * 每分钟执行一次
@@ -42,11 +50,7 @@ public class LiveCompletionPointsTask {
      */
     @Scheduled(cron = "0 */1 * * * ?")
     public void checkCompletionStatus() {
-        if (saasTaskEnabled && !TenantTaskRunner.isInTenantExecution()) {
-            tenantTaskRunner.runForEachTenant("checkCompletionStatus", this::doCheckCompletionStatus);
-            return;
-        }
-        doCheckCompletionStatus();
+        runWithSaaSTenant("checkCompletionStatus", this::doCheckCompletionStatus);
     }
 
     private void doCheckCompletionStatus() {

fs-live-app/src/main/java/com/fs/live/task/Task.java

@@ -89,14 +89,23 @@ public class Task {
     @Value("${saas.task.enabled}")
     private boolean saasTaskEnabled;
 
-    @Scheduled(cron = "0 0/1 * * * ?")
-    @DistributeLock(key = "updateLiveStatusByTime", scene = "task")
-    public void updateLiveStatusByTime() {
+    /**
+     * SaaS 多租户定时任务执行模板方法。
+     * 如果启用了 SaaS 且非租户执行上下文中，则按租户逐个执行；
+     * 否则直接执行（兼容单库模式或被 TenantTaskRunner 调用时）。
+     */
+    private void runWithSaaSTenant(String taskName, Runnable action) {
         if (saasTaskEnabled && !TenantTaskRunner.isInTenantExecution()) {
-            tenantTaskRunner.runForEachTenant("updateLiveStatusByTime", this::doUpdateLiveStatusByTime);
+            tenantTaskRunner.runForEachTenant(taskName, action);
             return;
         }
-        doUpdateLiveStatusByTime();
+        action.run();
+    }
+
+    @Scheduled(cron = "0 0/1 * * * ?")
+    @DistributeLock(key = "updateLiveStatusByTime", scene = "task")
+    public void updateLiveStatusByTime() {
+        runWithSaaSTenant("updateLiveStatusByTime", this::doUpdateLiveStatusByTime);
     }
 
     private void doUpdateLiveStatusByTime() {
@@ -256,11 +265,7 @@ public class Task {
     @Scheduled(cron = "0/1 * * * * ?")
     @DistributeLock(key = "liveLotteryTask", scene = "task")
     public void liveLotteryTask() {
-        if (saasTaskEnabled && !TenantTaskRunner.isInTenantExecution()) {
-            tenantTaskRunner.runForEachTenant("liveLotteryTask", this::doLiveLotteryTask);
-            return;
-        }
-        doLiveLotteryTask();
+        runWithSaaSTenant("liveLotteryTask", this::doLiveLotteryTask);
     }
 
     private void doLiveLotteryTask() {
@@ -394,11 +399,7 @@ public class Task {
     @Scheduled(cron = "0/1 * * * * ?")
     @DistributeLock(key = "liveAutoTask", scene = "task")
     public void liveAutoTask() {
-        if (saasTaskEnabled && !TenantTaskRunner.isInTenantExecution()) {
-            tenantTaskRunner.runForEachTenant("liveAutoTask", this::doLiveAutoTask);
-            return;
-        }
-        doLiveAutoTask();
+        runWithSaaSTenant("liveAutoTask", this::doLiveAutoTask);
     }
 
     private void doLiveAutoTask() {
@@ -435,11 +436,7 @@ public class Task {
     @DistributeLock(key = "autoUpdateWatchReward", scene = "task")
     @Transactional
     public void autoUpdateWatchReward() {
-        if (saasTaskEnabled && !TenantTaskRunner.isInTenantExecution()) {
-            tenantTaskRunner.runForEachTenant("autoUpdateWatchReward", this::doAutoUpdateWatchReward);
-            return;
-        }
-        doAutoUpdateWatchReward();
+        runWithSaaSTenant("autoUpdateWatchReward", this::doAutoUpdateWatchReward);
     }
 
     private void doAutoUpdateWatchReward() {
@@ -548,11 +545,7 @@ public class Task {
      */
     @Scheduled(cron = "0 0/1 * * * ?")// 每分钟执行一次
     public void syncLiveDataToDB() {
-        if (saasTaskEnabled && !TenantTaskRunner.isInTenantExecution()) {
-            tenantTaskRunner.runForEachTenant("syncLiveDataToDB", this::doSyncLiveDataToDB);
-            return;
-        }
-        doSyncLiveDataToDB();
+        runWithSaaSTenant("syncLiveDataToDB", this::doSyncLiveDataToDB);
     }
 
     private void doSyncLiveDataToDB() {
@@ -677,11 +670,7 @@ public class Task {
     @Scheduled(cron = "0/5 * * * * ?")
     @DistributeLock(key = "updateRedQuantityNum", scene = "task")
     public void updateRedQuantityNum() {
-        if (saasTaskEnabled && !TenantTaskRunner.isInTenantExecution()) {
-            tenantTaskRunner.runForEachTenant("updateRedQuantityNum", this::doUpdateRedQuantityNum);
-            return;
-        }
-        doUpdateRedQuantityNum();
+        runWithSaaSTenant("updateRedQuantityNum", this::doUpdateRedQuantityNum);
     }
 
     private void doUpdateRedQuantityNum() {
@@ -695,11 +684,7 @@ public class Task {
     @Scheduled(cron = "0/10 * * * * ?")
     @DistributeLock(key = "scanLiveTagMark", scene = "task")
     public void scanLiveTagMark() {
-        if (saasTaskEnabled && !TenantTaskRunner.isInTenantExecution()) {
-            tenantTaskRunner.runForEachTenant("scanLiveTagMark", this::doScanLiveTagMark);
-            return;
-        }
-        doScanLiveTagMark();
+        runWithSaaSTenant("scanLiveTagMark", this::doScanLiveTagMark);
     }
 
     private void doScanLiveTagMark() {
@@ -875,11 +860,7 @@ public class Task {
     @Scheduled(cron = "0/30 * * * * ?")
     @DistributeLock(key = "scanLiveWatchUserStatus", scene = "task")
     public void scanLiveWatchUserStatus() {
-        if (saasTaskEnabled && !TenantTaskRunner.isInTenantExecution()) {
-            tenantTaskRunner.runForEachTenant("scanLiveWatchUserStatus", this::doScanLiveWatchUserStatus);
-            return;
-        }
-        doScanLiveWatchUserStatus();
+        runWithSaaSTenant("scanLiveWatchUserStatus", this::doScanLiveWatchUserStatus);
     }
 
     private void doScanLiveWatchUserStatus() {
@@ -1058,11 +1039,7 @@ public class Task {
     @Scheduled(cron = "0 0/1 * * * ?")
     @DistributeLock(key = "updateLiveWatchUserStatus", scene = "task")
     public void updateLiveWatchUserStatus() {
-        if (saasTaskEnabled && !TenantTaskRunner.isInTenantExecution()) {
-            tenantTaskRunner.runForEachTenant("updateLiveWatchUserStatus", this::doUpdateLiveWatchUserStatus);
-            return;
-        }
-        doUpdateLiveWatchUserStatus();
+        runWithSaaSTenant("updateLiveWatchUserStatus", this::doUpdateLiveWatchUserStatus);
     }
 
     private void doUpdateLiveWatchUserStatus() {

fs-live-app/src/main/java/com/fs/live/task/TenantTaskRunner.java

@@ -18,9 +18,12 @@ import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
+import java.sql.Connection;
+import java.sql.ResultSet;
 import java.util.Collections;
 import java.util.Date;
 import java.util.List;
+import java.util.concurrent.*;
 import java.util.stream.Collectors;
 
 /**
@@ -34,6 +37,16 @@ public class TenantTaskRunner {
     /** ThreadLocal 防止任务方法递归二次分发 */
     private static final ThreadLocal<Boolean> IN_TENANT_EXECUTION = ThreadLocal.withInitial(() -> false);
 
+    /** 租户并行执行线程池（守护线程，应用关闭时自动终止） */
+    private final ExecutorService tenantExecutor = Executors.newFixedThreadPool(
+            Runtime.getRuntime().availableProcessors(),
+            r -> {
+                Thread t = new Thread(r, "saas-tenant-task");
+                t.setDaemon(true);
+                return t;
+            }
+    );
+
     @Resource
     private TenantDataSourceManager tenantDataSourceManager;
     @Resource
@@ -41,12 +54,14 @@ public class TenantTaskRunner {
     @Resource
     private SysConfigMapper sysConfigMapper;
 
+
+
     public static boolean isInTenantExecution() {
         return Boolean.TRUE.equals(IN_TENANT_EXECUTION.get());
     }
 
     /**
-     * 对每个启用且未过期的租户执行无参逻辑（任务名用于日志）。
+     * 对每个启用且未过期的租户并行执行无参逻辑（任务名用于日志）。
      */
     public void runForEachTenant(String taskName, Runnable action) {
         List<TenantInfo> tenants = getValidTenants();
@@ -54,8 +69,24 @@ public class TenantTaskRunner {
             log.debug("[SaaS Live Task] 无有效租户，跳过任务: {}", taskName);
             return;
         }
+        // 并行提交所有租户任务，等待全部完成
+        List<Future<?>> futures = new java.util.ArrayList<>(tenants.size());
         for (TenantInfo tenant : tenants) {
-            runForOneTenant(tenant, taskName, action);
+            futures.add(tenantExecutor.submit(() -> runForOneTenant(tenant, taskName, action)));
+        }
+        // 等待全部完成，收集异常
+        for (int i = 0; i < futures.size(); i++) {
+            try {
+                futures.get(i).get(5, TimeUnit.MINUTES);
+            } catch (TimeoutException e) {
+                log.error("[SaaS Live Task] 租户 {} 执行任务 {} 超时", tenants.get(i).getTenantCode(), taskName);
+            } catch (ExecutionException e) {
+                log.error("[SaaS Live Task] 租户 {} 执行任务 {} 异常", tenants.get(i).getTenantCode(), taskName, e.getCause());
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.warn("[SaaS Live Task] 任务 {} 被中断", taskName);
+                break;
+            }
         }
     }
 
@@ -85,6 +116,13 @@ public class TenantTaskRunner {
             log.info("[SaaS Live Task] 切换数据源 dataSource={}, tenantId={}, tenantCode={}, task={}",
                     dsKey, tenant.getId(), tenant.getTenantCode(), taskName);
 
+            // 检查租户是否开通了直播功能（live 表是否有数据）
+            if (!isLiveModuleEnabled(tenant)) {
+                log.error("[SaaS Live Task] 租户 tenantCode={} 未开通直播功能，跳过任务: {}",
+                        tenant.getTenantCode(), taskName);
+                return;
+            }
+
             // 加载租户项目配置
             SysConfig cfg = sysConfigMapper.selectConfigByConfigKey("projectConfig");
             if (cfg != null && StringUtils.isNotBlank(cfg.getConfigValue())) {
@@ -118,4 +156,23 @@ public class TenantTaskRunner {
             SecurityContextHolder.clearContext();
         }
     }
+
+    /**
+     * 检查租户库中 live 表是否有数据，判断是否开通了直播功能。
+     * 直接使用租户数据源，不依赖 ThreadLocal 路由。
+     */
+    private boolean isLiveModuleEnabled(TenantInfo tenant) {
+        try {
+            javax.sql.DataSource ds = tenantDataSourceManager.getTenantDataSource(tenant.getId());
+            try (Connection conn = ds.getConnection();
+                 java.sql.Statement stmt = conn.createStatement();
+                 ResultSet rs = stmt.executeQuery("SELECT 1 FROM live LIMIT 1")) {
+                return rs.next();
+            }
+        } catch (Exception e) {
+            log.error("[SaaS Live Task] 租户 tenantCode={} 检查直播模块失败: {}",
+                    tenant.getTenantCode(), e.getMessage());
+            return false;
+        }
+    }
 }

fs-live-app/src/main/java/com/fs/live/websocket/auth/WebSocketConfigurator.java

@@ -79,8 +79,13 @@ public class WebSocketConfigurator extends ServerEndpointConfig.Configurator {
         }
 
         // SaaS：握手阶段查库，将 tenantId/dsKey 存入 userProperties，后续 onOpen/onMessage/onClose 无需再查库
-        TenantChannelContext tenantChannelContext = SpringUtils.getBean(TenantChannelContext.class);
-        tenantChannelContext.resolveAndStore((String) userProperties.get(AttrConstant.TENANT_CODE), userProperties);
+        try {
+            TenantChannelContext tenantChannelContext = SpringUtils.getBean(TenantChannelContext.class);
+            tenantChannelContext.resolveAndStore(tenantCode, userProperties);
+        } catch (Exception e) {
+            log.error("[SaaS WS] 握手阶段绑定租户失败, tenantCode={}", tenantCode, e);
+            throw new BaseException("租户信息验证失败: " + tenantCode);
+        }
 
         // 验证token
         if (parameterMap.containsKey(tokenKey)) {

fs-user-app/src/main/java/com/fs/FsUserAppApplication.java

@@ -15,7 +15,7 @@ import org.springframework.transaction.annotation.EnableTransactionManagement;
 @SpringBootApplication(exclude = { DataSourceAutoConfiguration.class })
 @EnableTransactionManagement
 @EnableSwaggerBootstrapUI
-//@EnableScheduling
+@EnableScheduling
 public class FsUserAppApplication
 {
     public static void main(String[] args)