1、处理直播当场直播留存问题

2、处理发送红包无法领取问题（权限未申请）
3、避免唯一键多次进入问题

fs-live-app/src/main/java/com/fs/live/redis/LiveWsBroadcastSubscriber.java

@@ -1,5 +1,6 @@
 package com.fs.live.redis;
 
+import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.fs.common.constant.LiveKeysConstant;
 import com.fs.common.utils.StringUtils;
@@ -12,6 +13,7 @@ import org.springframework.data.redis.connection.MessageListener;
 import org.springframework.stereotype.Component;
 
 import java.nio.charset.StandardCharsets;
+import java.util.Map;
 
 /**
  * 订阅 admin 等服务发布的直播 WebSocket 广播，推送给 App 端
@@ -30,7 +32,11 @@ public class LiveWsBroadcastSubscriber implements MessageListener {
             return;
         }
         try {
-            JSONObject payload = JSONObject.parseObject(body);
+            JSONObject payload = parsePayload(body);
+            if (payload == null) {
+                log.warn("[LiveWsBroadcast] 忽略无效消息: {}", body);
+                return;
+            }
             Long liveId = payload.getLong("liveId");
             String cmd = payload.getString("cmd");
             if (liveId == null || StringUtils.isEmpty(cmd)) {
@@ -69,7 +75,31 @@ public class LiveWsBroadcastSubscriber implements MessageListener {
         }
     }
 
+    /**
+     * 兼容 RedisTemplate + FastJson WriteClassName 发布 String 时多包一层引号的情况，
+     * 仅做解析兼容，不改变后续广播逻辑。
+     */
+    @SuppressWarnings("unchecked")
+    private JSONObject parsePayload(String body) {
+        Object parsed = JSON.parse(body);
+        if (parsed instanceof String) {
+            parsed = JSON.parse((String) parsed);
+        }
+        if (parsed instanceof JSONObject) {
+            return (JSONObject) parsed;
+        }
+        if (parsed instanceof Map) {
+            return new JSONObject((Map<String, Object>) parsed);
+        }
+        // 明文 JSON 对象兜底
+        try {
+            return JSONObject.parseObject(body);
+        } catch (Exception e) {
+            return null;
+        }
+    }
+
     public static String channel() {
         return LiveKeysConstant.LIVE_WS_BROADCAST_CHANNEL;
     }
-}
+}

fs-service/src/main/java/com/fs/live/mapper/LiveCompletionPointsRecordMapper.java

@@ -29,10 +29,11 @@ public interface LiveCompletionPointsRecordMapper {
     int reuseRecordForNewSession(LiveCompletionPointsRecord record);
 
     /**
-     * 查询用户某天的完课记录
+     * 查询用户某天的完课记录（走主库，避免从库延迟导致误判后撞唯一键）
      */
-    LiveCompletionPointsRecord selectByUserAndDate(@Param("liveId") Long liveId, 
-                                                     @Param("userId") Long userId, 
+    @DataSource(DataSourceType.MASTER)
+    LiveCompletionPointsRecord selectByUserAndDate(@Param("liveId") Long liveId,
+                                                     @Param("userId") Long userId,
                                                      @Param("currentDate") Date currentDate);
 
     /**

fs-service/src/main/java/com/fs/live/service/impl/LiveCompletionPointsRecordServiceImpl.java

@@ -16,6 +16,8 @@ import com.fs.live.service.ILiveWatchUserService;
 import com.fs.live.utils.LiveCompletionConfigUtils;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.dao.DuplicateKeyException;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -119,34 +121,14 @@ public class LiveCompletionPointsRecordServiceImpl implements ILiveCompletionPoi
             }
 
             LocalDate today = LocalDate.now();
-            Date currentDate = Date.from(today.atStartOfDay(ZoneId.systemDefault()).toInstant());
+            // 使用 java.sql.Date，避免 Timestamp 时区导致 DATE 等值匹配失败
+            Date currentDate = java.sql.Date.valueOf(today);
 
             // 7. 按唯一键 uk_live_user_date（live+user+当天）先查主库当天记录，避免从库延迟/并发重复插入
             LiveCompletionPointsRecord todayRecord = recordMapper.selectByUserAndDate(liveId, userId, currentDate);
             if (todayRecord != null) {
-                if (isRecordInCurrentSession(todayRecord, live.getStartTime())) {
-                    // 本场已创建过，不再插入
-                    return null;
-                }
-                // 同日上一场残留：复用更新，避开唯一键冲突，重开后可再领
-                LiveCompletionPointsRecord latestRecord = recordMapper.selectLatestByUser(userId);
-                int continuousDays = calcContinuousDays(latestRecord, today);
-                int points = calculatePoints(continuousDays, pointsConfig);
-
-                todayRecord.setWatchDuration(actualWatchDuration);
-                todayRecord.setVideoDuration(videoDuration);
-                todayRecord.setCompletionRate(watchRate);
-                todayRecord.setContinuousDays(continuousDays);
-                todayRecord.setPointsAwarded(points);
-                if (latestRecord != null && !Objects.equals(latestRecord.getId(), todayRecord.getId())) {
-                    todayRecord.setLastCompletionDate(latestRecord.getCurrentCompletionDate());
-                }
-                recordMapper.reuseRecordForNewSession(todayRecord);
-                todayRecord.setReceiveStatus(0);
-                todayRecord.setReceiveTime(null);
-                todayRecord.setCreateTime(new Date());
-                log.info("同日重开复用完课记录, liveId={}, userId={}, recordId={}", liveId, userId, todayRecord.getId());
-                return todayRecord;
+                return handleExistingTodayRecord(todayRecord, live, userId, today,
+                        actualWatchDuration, videoDuration, watchRate, pointsConfig);
             }
 
             // 8. 查询最近一次完课记录（不限直播间）,计算连续天数
@@ -174,23 +156,90 @@ public class LiveCompletionPointsRecordServiceImpl implements ILiveCompletionPoi
 
             try {
                 recordMapper.insertRecord(record);
-            } catch (org.springframework.dao.DuplicateKeyException e) {
-                // 实时推送与定时任务并发时可能撞唯一键，视为已存在
-                log.info("完课记录已存在(并发), liveId={}, userId={}", liveId, userId);
-                return null;
+                return record;
+            } catch (Exception e) {
+                if (!isDuplicateKey(e)) {
+                    throw e;
+                }
+                // 并发/时区漏查后撞唯一键：回查当天记录，同日重开则复用
+                log.info("完课记录插入撞唯一键, liveId={}, userId={}, msg={}", liveId, userId, e.getMessage());
+                LiveCompletionPointsRecord existed = recordMapper.selectByUserAndDate(liveId, userId, currentDate);
+                if (existed == null) {
+                    return null;
+                }
+                return handleExistingTodayRecord(existed, live, userId, today,
+                        actualWatchDuration, videoDuration, watchRate, pointsConfig);
             }
 
-            return record;
-
-        } catch (org.springframework.dao.DuplicateKeyException e) {
-            log.info("完课记录已存在(并发), liveId={}, userId={}", liveId, userId);
-            return null;
         } catch (Exception e) {
+            if (isDuplicateKey(e)) {
+                log.info("完课记录已存在(并发), liveId={}, userId={}", liveId, userId);
+                return null;
+            }
             log.error("检查并创建完课记录失败, liveId={}, userId={}", liveId, userId, e);
             throw e;
         }
     }
 
+    /**
+     * 处理当天已存在记录：本场已有则跳过；同日上一场则复用更新并重置为可再领。
+     */
+    private LiveCompletionPointsRecord handleExistingTodayRecord(LiveCompletionPointsRecord todayRecord,
+                                                                 Live live,
+                                                                 Long userId,
+                                                                 LocalDate today,
+                                                                 Long actualWatchDuration,
+                                                                 Long videoDuration,
+                                                                 BigDecimal watchRate,
+                                                                 int[] pointsConfig) {
+        if (isRecordInCurrentSession(todayRecord, live.getStartTime())) {
+            // 本场已创建过，不再插入
+            return null;
+        }
+        // 同日上一场残留：复用更新，避开唯一键冲突，重开后可再领
+        LiveCompletionPointsRecord latestRecord = recordMapper.selectLatestByUser(userId);
+        int continuousDays = calcContinuousDays(latestRecord, today);
+        int points = calculatePoints(continuousDays, pointsConfig);
+
+        todayRecord.setWatchDuration(actualWatchDuration);
+        todayRecord.setVideoDuration(videoDuration);
+        todayRecord.setCompletionRate(watchRate);
+        todayRecord.setContinuousDays(continuousDays);
+        todayRecord.setPointsAwarded(points);
+        if (latestRecord != null && !Objects.equals(latestRecord.getId(), todayRecord.getId())) {
+            todayRecord.setLastCompletionDate(latestRecord.getCurrentCompletionDate());
+        }
+        recordMapper.reuseRecordForNewSession(todayRecord);
+        todayRecord.setReceiveStatus(0);
+        todayRecord.setReceiveTime(null);
+        todayRecord.setCreateTime(new Date());
+        log.info("同日重开复用完课记录, liveId={}, userId={}, recordId={}", live.getLiveId(), userId, todayRecord.getId());
+        return todayRecord;
+    }
+
+    /** 是否唯一键冲突（兼容 MyBatis PersistenceException 包装） */
+    private boolean isDuplicateKey(Throwable e) {
+        Throwable t = e;
+        while (t != null) {
+            if (t instanceof DuplicateKeyException
+                    || t instanceof java.sql.SQLIntegrityConstraintViolationException) {
+                return true;
+            }
+            if (t instanceof DataIntegrityViolationException) {
+                String msg = t.getMessage();
+                if (msg != null && (msg.contains("Duplicate") || msg.contains("uk_live_user_date"))) {
+                    return true;
+                }
+            }
+            String msg = t.getMessage();
+            if (msg != null && (msg.contains("Duplicate entry") || msg.contains("uk_live_user_date"))) {
+                return true;
+            }
+            t = t.getCause();
+        }
+        return false;
+    }
+
     /** 记录是否属于当前开播场次（create_time >= startTime） */
     private boolean isRecordInCurrentSession(LiveCompletionPointsRecord record, LocalDateTime startTime) {
         if (record == null) {

fs-service/src/main/resources/application-druid-tyt-test.yml

@@ -247,4 +247,4 @@ wechat:
         redirectUri: https://admin.tyt.com/prod-api/callback
         isNeedScan: false
 
-
+enableRedPackAccount: 1

fs-service/src/main/resources/application-druid-tyt.yml

@@ -247,4 +247,5 @@ wechat:
         redirectUri: https://admin.tyt.com/prod-api/callback
         isNeedScan: false
 
+enableRedPackAccount: 1
 

fs-service/src/main/resources/mapper/live/LiveCompletionPointsRecordMapper.xml

@@ -88,12 +88,12 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         WHERE id = #{id}
     </update>
 
-    <!-- 查询用户某天的完课记录 -->
+    <!-- 查询用户某天的完课记录（按日期比较，避免时区/时间戳导致等值匹配失败） -->
     <select id="selectByUserAndDate" resultMap="LiveCompletionPointsRecordResult">
         SELECT * FROM live_completion_points_record
         WHERE live_id = #{liveId}
           AND user_id = #{userId}
-          AND current_completion_date = #{currentDate}
+          AND DATE(current_completion_date) = DATE(#{currentDate})
         LIMIT 1
     </select>
 

fs-user-app/src/main/java/com/fs/app/controller/live/LiveController.java

@@ -31,6 +31,7 @@ import com.fs.live.mapper.LiveVideoMapper;
 import com.fs.live.mapper.LiveWatchUserMapper;
 import com.fs.live.param.FsLiveEncryptLinkParam;
 import com.fs.live.param.LiveNotifyParam;
+import com.fs.live.param.SignPO;
 import com.fs.live.service.*;
 import com.fs.live.utils.LiveCompletionConfigUtils;
 import com.fs.live.vo.LiveVo;
@@ -429,10 +430,10 @@ public class LiveController extends AppBaseController {
 	 * 直播签到并领取该次配置奖励（积分红包/优惠券/金额红包）。
 	 * 金额红包成功时响应含 needWxConfirm、package 等，App 需跳转微信确认收款。
 	 */
-	@Login
+//	@Login
 	@ApiOperation("直播签到领取奖励")
 	@PostMapping("/sign")
-	public R sign(@RequestBody com.fs.live.param.SignPO sign) {
+	public R sign(@RequestBody SignPO sign) {
 		try {
 			Long userId = Long.valueOf(getUserId());
 			return liveFacadeService.signClaim(sign, userId);

fs-user-app/src/main/java/com/fs/app/facade/impl/LiveFacadeServiceImpl.java

@@ -4,6 +4,8 @@ import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.json.JSONUtil;
 import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 import com.fs.app.facade.LiveFacadeService;
 import com.fs.common.constant.LiveKeysConstant;
 import com.fs.common.core.controller.BaseController;
@@ -12,6 +14,12 @@ import com.fs.common.core.page.PageRequest;
 import com.fs.common.core.page.TableDataInfo;
 import com.fs.common.core.redis.RedisCache;
 import com.fs.common.utils.StringUtils;
+import com.fs.company.domain.CompanyUser;
+import com.fs.company.service.ICompanyUserService;
+import com.fs.his.domain.FsUser;
+import com.fs.his.param.WxSendRedPacketParam;
+import com.fs.his.service.IFsStorePaymentService;
+import com.fs.his.service.IFsUserService;
 import com.fs.live.domain.*;
 import com.fs.framework.aspectj.lock.DistributeLock;
 import com.fs.live.param.CouponPO;
@@ -25,6 +33,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.math.BigDecimal;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.time.temporal.ChronoUnit;
@@ -74,10 +83,13 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
     private ILiveCouponIssueService liveCouponIssueService;
 
     @Autowired
-    private com.fs.his.service.IFsUserService fsUserService;
+    private IFsUserService fsUserService;
 
     @Autowired
-    private com.fs.his.service.IFsStorePaymentService paymentService;
+    private IFsStorePaymentService paymentService;
+
+    @Autowired
+    private ICompanyUserService companyUserService;
 
     @Autowired
     private ILiveRewardRecordService liveRewardRecordService;
@@ -308,6 +320,7 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
 
     /**
      * 直播签到领取：写 live_sign_record 防重，再按 rewards 发奖并绑定留存。
+     * 防重按「本场开播」判定（createTime >= live.startTime），结束后重开可再签。
      * 金额红包成功时返回 package 等字段，供 App 跳转微信确认收款。
      */
     @Override
@@ -319,18 +332,23 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
         Long liveId = sign.getLiveId();
         String signNo = String.valueOf(sign.getSignNo());
 
-        // 防重复签到
+        Live live = liveService.selectLiveByLiveId(liveId);
+        if (live == null) {
+            return R.error("直播间不存在");
+        }
+
+        // 防重复签到：仅拦截本场（开播后）已签记录，结束后重开可再签
         LiveSignRecord existQuery = new LiveSignRecord();
         existQuery.setLiveId(liveId);
         existQuery.setUserId(userId);
         existQuery.setSignNo(signNo);
         List<LiveSignRecord> exists = liveSignRecordService.selectLiveSignRecordList(existQuery);
-        if (CollUtil.isNotEmpty(exists)) {
+        if (hasSignedInCurrentSession(exists, live.getStartTime())) {
             return R.error("您已完成该次签到");
         }
 
         // 从触发缓存取奖励留存；没有则从已执行签到任务解析
-        List<com.alibaba.fastjson.JSONObject> rewards = resolveSignRewards(liveId, signNo, sign.getTaskId());
+        List<JSONObject> rewards = resolveSignRewards(liveId, signNo, sign.getTaskId());
         if (CollUtil.isEmpty(rewards)) {
             return R.error("签到任务尚未开启或奖励配置不存在");
         }
@@ -344,7 +362,7 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
         liveSignRecordService.save(record);
 
         List<Map<String, Object>> claimResults = new ArrayList<>();
-        for (com.alibaba.fastjson.JSONObject item : rewards) {
+        for (JSONObject item : rewards) {
             Long rewardType = item.getLong("rewardType");
             Long opLogId = item.getLong("opLogId");
             Map<String, Object> one = new HashMap<>();
@@ -422,9 +440,10 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
     /**
      * 签到金额红包：发起微信商家转账（金额精确到 0.1 元）。
      * 成功时返回 package/mchId/appId/orderCode，供 App 跳转微信确认收款。
+     * 扣款公司取用户绑定销售（bindCompanyUserId）所属公司，而非直播间公司。
      */
-    private R claimSignCashRed(Long liveId, Long userId, com.alibaba.fastjson.JSONObject item, SignPO sign) {
-        java.math.BigDecimal amount = item.getBigDecimal("amount");
+    private R claimSignCashRed(Long liveId, Long userId, JSONObject item, SignPO sign) {
+        BigDecimal amount = item.getBigDecimal("amount");
         if (amount == null || amount.compareTo(java.math.BigDecimal.ZERO) <= 0) {
             return R.error("金额红包配置无效");
         }
@@ -432,13 +451,20 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
         if (live == null) {
             return R.error("直播间不存在");
         }
-        com.fs.his.domain.FsUser user = fsUserService.selectFsUserById(userId);
+        FsUser user = fsUserService.selectFsUserById(userId);
         if (user == null) {
             return R.error("用户不存在");
         }
-        if (live.getCompanyId() == null) {
-            return R.error("直播间未绑定公司，无法发放金额红包");
+        // 通过用户绑定销售ID查询销售所属公司，作为红包扣款公司
+        Long bindCompanyUserId = user.getBindCompanyUserId();
+        if (bindCompanyUserId == null) {
+            return R.error("用户未绑定销售，无法发放金额红包");
+        }
+        CompanyUser companyUser = companyUserService.selectCompanyUserById(bindCompanyUserId);
+        if (companyUser == null || companyUser.getCompanyId() == null) {
+            return R.error("绑定销售不存在或未归属公司，无法发放金额红包");
         }
+        Long companyId = companyUser.getCompanyId();
         String openId = null;
         if (sign != null && sign.getSource() != null && sign.getSource() == 2
                 && StringUtils.isNotEmpty(user.getCourseMaOpenId())) {
@@ -451,16 +477,16 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
         if (StringUtils.isEmpty(openId)) {
             return R.error("请使用微信登录后再领取金额红包");
         }
-        com.fs.his.param.WxSendRedPacketParam packetParam = new com.fs.his.param.WxSendRedPacketParam();
+        WxSendRedPacketParam packetParam = new com.fs.his.param.WxSendRedPacketParam();
         packetParam.setOpenId(openId);
         packetParam.setAmount(amount.setScale(1, java.math.RoundingMode.HALF_UP));
-        packetParam.setCompanyId(live.getCompanyId());
+        packetParam.setCompanyId(companyId);
         packetParam.setSource(sign != null && sign.getSource() != null ? sign.getSource() : 1);
         packetParam.setAppId(sign != null ? sign.getAppId() : null);
         packetParam.setRedPacketMode(1);
         packetParam.setUser(user);
         try {
-            R sendResult = paymentService.sendRedPacket(packetParam);
+            R sendResult = paymentService.sendAppRedPacket(packetParam);
             if (!Integer.valueOf(200).equals(sendResult.get("code"))) {
                 return sendResult;
             }
@@ -473,7 +499,7 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
                 rewardRecord.setNum(amount);
                 rewardRecord.setIncomeType(1L);
                 rewardRecord.setSourceType(7L);
-                rewardRecord.setSourceId(live.getCompanyId() != null ? live.getCompanyId() : 0L);
+                rewardRecord.setSourceId(companyId);
                 rewardRecord.setCreateBy(String.valueOf(userId));
                 liveRewardRecordService.insertLiveRewardRecord(rewardRecord);
             } catch (Exception e) {
@@ -489,14 +515,14 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
     /**
      * 解析本次签到可领取奖励：优先读触发后 Redis 缓存（含 opLogId），否则回落已执行签到任务 content.rewards。
      */
-    private List<com.alibaba.fastjson.JSONObject> resolveSignRewards(Long liveId, String signNo, Long taskId) {
-        List<com.alibaba.fastjson.JSONObject> rewards = new ArrayList<>();
+    private List<JSONObject> resolveSignRewards(Long liveId, String signNo, Long taskId) {
+        List<JSONObject> rewards = new ArrayList<>();
         try {
             Object cache = redisCache.getCacheObject(String.format(LiveKeysConstant.LIVE_SIGN_OP_LOGS, liveId, signNo));
             if (cache != null) {
-                com.alibaba.fastjson.JSONObject data = JSON.parseObject(String.valueOf(cache));
+                JSONObject data = JSON.parseObject(String.valueOf(cache));
                 if (data != null && data.getJSONArray("rewards") != null) {
-                    com.alibaba.fastjson.JSONArray arr = data.getJSONArray("rewards");
+                    JSONArray arr = data.getJSONArray("rewards");
                     for (int i = 0; i < arr.size(); i++) {
                         rewards.add(arr.getJSONObject(i));
                     }
@@ -525,7 +551,7 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
                         continue;
                     }
                     try {
-                        com.alibaba.fastjson.JSONObject c = JSON.parseObject(t.getContent());
+                        JSONObject c = JSON.parseObject(t.getContent());
                         if (c != null && signNo.equals(String.valueOf(c.get("signNo")))) {
                             task = t;
                             break;
@@ -539,14 +565,14 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
             return rewards;
         }
         try {
-            com.alibaba.fastjson.JSONObject content = JSON.parseObject(task.getContent());
-            com.alibaba.fastjson.JSONArray arr = content.getJSONArray("rewards");
+            JSONObject content = JSON.parseObject(task.getContent());
+            JSONArray arr = content.getJSONArray("rewards");
             if (arr != null) {
                 for (int i = 0; i < arr.size(); i++) {
                     rewards.add(arr.getJSONObject(i));
                 }
             } else if (content.getLong("rewardType") != null) {
-                com.alibaba.fastjson.JSONObject single = new com.alibaba.fastjson.JSONObject();
+                JSONObject single = new JSONObject();
                 single.put("rewardType", content.getLong("rewardType"));
                 single.put("rewardId", content.getLong("rewardId"));
                 single.put("reward", content.get("reward"));
@@ -558,6 +584,26 @@ public class LiveFacadeServiceImpl extends BaseController implements LiveFacadeS
         return rewards;
     }
 
+    /**
+     * 是否已在当前开播场次签过该次签到。
+     * 以 live.startTime 为场次边界：结束后重开（startTime 更新）可再签。
+     */
+    private boolean hasSignedInCurrentSession(List<LiveSignRecord> records, LocalDateTime startTime) {
+        if (CollUtil.isEmpty(records)) {
+            return false;
+        }
+        if (startTime == null) {
+            return true;
+        }
+        return records.stream().anyMatch(r -> {
+            if (r.getCreateTime() == null) {
+                return true;
+            }
+            LocalDateTime recordLdt = r.getCreateTime().toInstant().atZone(ZONE_ID).toLocalDateTime();
+            return !recordLdt.isBefore(startTime);
+        });
+    }
+
     private Long resolveLiveDuration(Long liveId) {
         if (liveId == null) {
             return 0L;