package douyin import ( "errors" "bindbox-game/internal/pkg/logger" "bindbox-game/internal/repository/mysql" "bindbox-game/internal/repository/mysql/model" "bindbox-game/internal/service/game" "bindbox-game/internal/service/sysconfig" "context" "fmt" "strconv" "time" "go.uber.org/zap" "bindbox-game/internal/service/user" "gorm.io/gorm" ) // StartDouyinOrderSync 启动抖店订单定时同步任务 func StartDouyinOrderSync(l logger.CustomLogger, repo mysql.Repo, syscfg sysconfig.Service, ticketSvc game.TicketService, userSvc user.Service, titleSvc TitleAssigner) { svc := New(l, repo, syscfg, ticketSvc, userSvc, titleSvc) go func() { // 初始等待30秒让服务完全启动 time.Sleep(30 * time.Second) // 创建多个定时器,分频执行不同任务 ticker5min := time.NewTicker(5 * time.Minute) // 直播奖品发放 ticker1h := time.NewTicker(1 * time.Hour) // 全量订单同步 ticker2h := time.NewTicker(2 * time.Hour) // 退款状态同步 defer ticker5min.Stop() defer ticker1h.Stop() defer ticker2h.Stop() // 首次立即执行一次全量同步 ctx := context.Background() firstRun := true if firstRun { l.Info("[抖店定时同步] 首次启动,执行全量同步 (48小时)") if res, err := svc.SyncAllOrders(ctx, 48*time.Hour, true); err != nil { l.Error("[定时同步] 首次全量同步失败", zap.Error(err)) } else { l.Info("[定时同步] 首次全量同步完成", zap.String("info", res.DebugInfo)) } firstRun = false } l.Info("[抖店定时同步] 定时任务已启动", zap.String("直播奖品", "每5分钟"), zap.String("订单同步", "每1小时"), zap.String("退款同步", "每2小时")) for { select { case <-ticker5min.C: // ========== 每 5 分钟: 自动发放直播间奖品 ========== // 不调用抖音 API,只处理本地数据 ctx := context.Background() l.Debug("[定时发放] 开始发放直播奖品") if err := svc.GrantLivestreamPrizes(ctx); err != nil { l.Error("[定时发放] 发放直播奖品失败", zap.Error(err)) } else { l.Debug("[定时发放] 发放直播奖品完成") } case <-ticker1h.C: // ========== 每 1 小时: 全量订单同步 ========== // 调用抖音 API,同步最近 1 小时的订单 ctx := context.Background() l.Info("[定时同步] 开始全量订单同步 (1小时)") if res, err := svc.SyncAllOrders(ctx, 1*time.Hour, true); err != nil { l.Error("[定时同步] 全量同步失败", zap.Error(err)) } else { l.Info("[定时同步] 全量同步完成", zap.String("info", res.DebugInfo)) } case <-ticker2h.C: // ========== 每 2 小时: 退款状态同步 ========== // 调用抖音 API,检查订单退款状态 ctx := context.Background() l.Info("[定时同步] 开始退款状态同步") if err := svc.SyncRefundStatus(ctx); err != nil { l.Error("[定时同步] 同步退款状态失败", zap.Error(err)) } else { l.Debug("[定时同步] 退款状态同步完成") } } } }() l.Info("[抖店定时同步] 定时任务已启动") } // GrantMinesweeperQualifications 自动补发扫雷资格 // 逻辑:遍历已绑定抖音的用户 -> 查找其未归属的订单 -> 关联订单 -> 补发资格 func (s *service) GrantMinesweeperQualifications(ctx context.Context) error { db := s.repo.GetDbW().WithContext(ctx) // 1. 查找所有已绑定抖音的用户 var users []model.Users if err := s.repo.GetDbR().Where("douyin_user_id != '' AND douyin_user_id IS NOT NULL").Find(&users).Error; err != nil { return err } for _, u := range users { // 1.1 检查是否在黑名单中 var blacklistCount int64 if err := s.repo.GetDbR().Table("douyin_blacklist").Where("douyin_user_id = ?", u.DouyinUserID).Count(&blacklistCount).Error; err == nil && blacklistCount > 0 { continue } // 2. 查找该抖音ID下未关联(local_user_id=0 or empty)的订单 var orders []model.DouyinOrders if err := db.Where("douyin_user_id = ? AND (local_user_id = '' OR local_user_id = '0')", u.DouyinUserID).Find(&orders).Error; err != nil { continue } for _, order := range orders { // 3. 关联订单到用户 if err := db.Model(&order).Update("local_user_id", strconv.FormatInt(u.ID, 10)).Error; err != nil { s.logger.Error("[自动补发] 关联订单失败", zap.String("order_id", order.ShopOrderID), zap.Error(err)) continue } // 4. 如果是已支付待发货的订单(2),且未发奖,则补发 if order.OrderStatus == 2 && order.RewardGranted == 0 { orderID := order.ID s.logger.Info("[自动补发] 开始补发扫雷资格", zap.Int64("user_id", u.ID), zap.String("shop_order_id", order.ShopOrderID)) // 调用发奖服务 count := int64(1) if order.ProductCount > 0 { count = int64(order.ProductCount) } s.logger.Info("[自动补发] 发放数量", zap.Int64("count", count)) if err := s.ticketSvc.GrantTicket(ctx, u.ID, "minesweeper", int(count), "douyin_order", orderID, "定时任务补发"); err == nil { db.Model(&order).Update("reward_granted", int32(count)) s.logger.Info("[自动补发] 补发成功", zap.String("shop_order_id", order.ShopOrderID)) } else { s.logger.Error("[自动补发] 补发失败", zap.Error(err)) } } } } return nil } // GrantLivestreamPrizes 自动发放直播间奖品 // 逻辑:扫描 livestream_draw_logs 中 is_granted=0 的记录 -> 找到对应 ProductID -> 发放商品 func (s *service) GrantLivestreamPrizes(ctx context.Context) error { db := s.repo.GetDbW().WithContext(ctx) // 1. 查找未发放的记录 var logs []model.LivestreamDrawLogs if err := db.Where("is_granted = 0").Find(&logs).Error; err != nil { return err } anchorCodes := s.resolveActivityAnchorCodes(ctx, logs) for _, log := range logs { // 必须要有对应的本地用户ID if log.LocalUserID == 0 { // 尝试从 douyin_orders 补全 user_id var order model.DouyinOrders if err := s.repo.GetDbR().Where("shop_order_id = ?", log.ShopOrderID).First(&order).Error; err == nil { if uid, _ := strconv.ParseInt(order.LocalUserID, 10, 64); uid > 0 { log.LocalUserID = uid db.Model(&log).Update("local_user_id", uid) } } } if log.LocalUserID == 0 { continue // 还没关联到用户,跳过 } if code := anchorCodes[log.ActivityID]; code != "" { s.bindAnchorInviterIfNeeded(ctx, log.LocalUserID, code) } // 2. 查奖品关联的 ProductID var prize model.LivestreamPrizes if err := s.repo.GetDbR().Where("id = ?", log.PrizeID).First(&prize).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { if log.ProductID > 0 { prize = model.LivestreamPrizes{ ID: log.PrizeID, Name: log.PrizeName, ProductID: log.ProductID, } s.logger.Warn("[自动发放] 奖品配置缺失,使用快照兜底", zap.Int64("prize_id", log.PrizeID), zap.Int64("product_id", log.ProductID), zap.Int64("log_id", log.ID)) } else { s.logger.Error("[自动发放] 奖品不存在且缺少快照", zap.Int64("prize_id", log.PrizeID), zap.Int64("log_id", log.ID)) continue } } else { s.logger.Error("[自动发放] 查询奖品失败", zap.Int64("prize_id", log.PrizeID), zap.Error(err)) continue } } if prize.ProductID == 0 { s.logger.Warn("[自动发放] 奖品未关联商品ID,跳过", zap.Int64("prize_id", log.PrizeID), zap.String("name", prize.Name)) continue } // 3. 发放商品 (使用 GrantReward 创建新订单发放) sourceType := int32(5) // 5 代表直播间 req := user.GrantRewardRequest{ ProductID: prize.ProductID, Quantity: 1, SourceType: &sourceType, Remark: fmt.Sprintf("直播间抽奖: %s (关联抖店订单: %s)", log.PrizeName, log.ShopOrderID), } s.logger.Info("[自动发放] 开始发放直播商品", zap.Int64("user_id", log.LocalUserID), zap.Int64("product_id", prize.ProductID), zap.String("prize", log.PrizeName), ) res, err := s.userSvc.GrantReward(ctx, log.LocalUserID, req) if err != nil { s.logger.Error("[自动发放] 发放失败", zap.Error(err)) s.logRewardResult(ctx, log.ShopOrderID, log.DouyinUserID, log.LocalUserID, fmt.Sprintf("%d", prize.ProductID), log.PrizeID, "auto", "failed", err.Error()) // 如果发放失败是库存原因等,可能需要告警。暂时不重试,等下个周期。 } else { // 4. 更新发放状态 db.Model(&log).Update("is_granted", 1) s.logger.Info("[自动发放] 发放成功", zap.Int64("log_id", log.ID)) s.logRewardResult(ctx, log.ShopOrderID, log.DouyinUserID, log.LocalUserID, fmt.Sprintf("%d", prize.ProductID), log.PrizeID, "auto", "success", "发放成功") // 5. 自动虚拟发货 (本地状态更新) // 直播间奖品通常为虚拟发货,直接标记为已消费/已发货 if res != nil && res.OrderID > 0 { updates := map[string]interface{}{ "is_consumed": 1, "updated_at": time.Now(), } if err := s.repo.GetDbW().WithContext(ctx).Model(&model.Orders{}).Where("id = ?", res.OrderID).Updates(updates).Error; err != nil { s.logger.Error("[自动发放] 更新订单状态失败", zap.Int64("order_id", res.OrderID), zap.Error(err)) } // 更新发货记录 shippingUpdates := map[string]interface{}{ "status": 2, // 已发货 "shipped_at": time.Now(), "updated_at": time.Now(), } if err := s.repo.GetDbW().WithContext(ctx).Model(&model.ShippingRecords{}).Where("order_id = ?", res.OrderID).Updates(shippingUpdates).Error; err != nil { s.logger.Error("[自动发放] 更新发货记录失败", zap.Int64("order_id", res.OrderID), zap.Error(err)) } else { s.logger.Info("[自动发放] 虚拟发货完成(本地)", zap.Int64("order_id", res.OrderID)) } } } } return nil } // SyncRefundStatus 同步退款状态 // 逻辑:检查 douyin_orders 的状态变更,如果订单已退款,则标记对应的 livestream_draw_logs func (s *service) SyncRefundStatus(ctx context.Context) error { db := s.repo.GetDbW().WithContext(ctx) // 1. 查找所有关联直播抽奖但尚未标记退款的记录 var logs []model.LivestreamDrawLogs if err := db.Where("is_refunded = 0 AND shop_order_id != ''").Find(&logs).Error; err != nil { return err } refundedCount := 0 for _, log := range logs { // 2. 查找对应的抖店订单 var order model.DouyinOrders if err := s.repo.GetDbR().Where("shop_order_id = ?", log.ShopOrderID).First(&order).Error; err != nil { continue // 找不到订单,跳过 } // 3. 检查订单状态:抖店状态 4=已关闭 (包含退款/取消等关闭情况) // 状态说明: 3=已发货, 4=已关闭, 5=已完成 if order.OrderStatus == 4 { db.Model(&log).Update("is_refunded", 1) refundedCount++ s.logger.Info("[退款同步] 标记退款记录", zap.Int64("draw_log_id", log.ID), zap.String("shop_order_id", log.ShopOrderID), zap.Int32("order_status", order.OrderStatus), ) // 4. 如果用户已关联,回收资产 if log.LocalUserID > 0 { s.reclaimLivestreamAssets(ctx, &log) } } } if refundedCount > 0 { s.logger.Info("[退款同步] 本次同步完成", zap.Int("refunded_count", refundedCount)) } return nil } // reclaimLivestreamAssets 回收直播间发放的资产 // 逻辑:查找该用户通过此抽奖获得的 user_inventory,作废或扣除积分 func (s *service) reclaimLivestreamAssets(ctx context.Context, log *model.LivestreamDrawLogs) { db := s.repo.GetDbW().WithContext(ctx) // 1. 查找关联的 user_inventory 记录 // 直播间奖品是通过 GrantReward 发放的,会创建一个新的本地订单 // 我们需要通过 remark 或其他方式找到关联的 inventory // 由于 GrantReward 会在 remark 中记录 shop_order_id,我们通过这个来查找 var inventories []model.UserInventory // 查找用户持有的、来自直播间的资产(通过 remark 包含 shop_order_id 来关联) searchPattern := "%" + log.ShopOrderID + "%" if err := db.Where("user_id = ? AND status IN (1, 3) AND remark LIKE ?", log.LocalUserID, searchPattern).Find(&inventories).Error; err != nil { s.logger.Error("[资产回收] 查询资产失败", zap.Error(err), zap.Int64("user_id", log.LocalUserID)) return } if len(inventories) == 0 { // 尝试通过 prize_id 和 product_id 关联查找 var prize model.LivestreamPrizes if err := s.repo.GetDbR().Where("id = ?", log.PrizeID).First(&prize).Error; err == nil && prize.ProductID > 0 { // 查找该用户最近获得的该商品的资产(时间相近) db.Where("user_id = ? AND product_id = ? AND status IN (1, 3) AND created_at >= ?", log.LocalUserID, prize.ProductID, log.CreatedAt.Add(-time.Hour)). Order("created_at DESC").Limit(1).Find(&inventories) } } if len(inventories) == 0 { s.logger.Warn("[资产回收] 未找到可回收资产", zap.Int64("user_id", log.LocalUserID), zap.String("shop_order_id", log.ShopOrderID), ) return } // 2. 回收资产 rate := int64(1) var cfg model.SystemConfigs if err := s.repo.GetDbR().Where("config_key = ?", "points_exchange_per_cent").First(&cfg).Error; err == nil { var rv int64 _, _ = fmt.Sscanf(cfg.ConfigValue, "%d", &rv) if rv > 0 { rate = rv } } for _, inv := range inventories { if inv.Status == 1 { // 状态1(持有):作废 db.Model(&inv).Updates(map[string]any{ "status": 2, "remark": inv.Remark + "|refund_reclaimed", }) s.logger.Info("[资产回收] 作废持有资产", zap.Int64("inventory_id", inv.ID), zap.Int64("user_id", inv.UserID), ) } else if inv.Status == 3 { // 状态3(已兑换/发货):扣除积分 pointsToDeduct := inv.ValueCents * rate if pointsToDeduct <= 0 { // 兼容历史数据,兜底回退商品价格 var product model.Products if err := s.repo.GetDbR().Where("id = ?", inv.ProductID).First(&product).Error; err == nil { pointsToDeduct = product.Price * rate } } if pointsToDeduct > 0 { _, consumed, err := s.userSvc.ConsumePointsForRefund(ctx, inv.UserID, pointsToDeduct, "user_inventory", fmt.Sprintf("%d", inv.ID), "直播退款回收已兑换资产") if err != nil { s.logger.Error("[资产回收] 扣除积分失败", zap.Error(err), zap.Int64("user_id", inv.UserID)) } if consumed < pointsToDeduct { // 积分不足,标记用户 s.logger.Warn("[资产回收] 用户积分不足", zap.Int64("user_id", inv.UserID), zap.Int64("needed", pointsToDeduct), zap.Int64("consumed", consumed), ) // 可选:加入黑名单 // db.Exec("UPDATE users SET status = 3 WHERE id = ?", inv.UserID) } } // 作废记录 db.Model(&inv).Updates(map[string]any{ "status": 2, "remark": inv.Remark + "|refund_reclaimed_points_deducted", }) s.logger.Info("[资产回收] 扣除积分并作废", zap.Int64("inventory_id", inv.ID), zap.Int64("user_id", inv.UserID), ) } } // 3. 恢复奖品库存 db.Exec("UPDATE livestream_prizes SET remaining = remaining + 1 WHERE id = ? AND remaining >= 0", log.PrizeID) s.logger.Info("[资产回收] 恢复奖品库存", zap.Int64("prize_id", log.PrizeID)) } func (s *service) resolveActivityAnchorCodes(ctx context.Context, logs []model.LivestreamDrawLogs) map[int64]string { result := make(map[int64]string) if len(logs) == 0 { return result } type anchorMeta struct { channelID int64 channelCode string } activityMeta := make(map[int64]anchorMeta) var activityIDs []int64 for _, log := range logs { if log.ActivityID <= 0 { continue } if _, exists := activityMeta[log.ActivityID]; exists { continue } activityMeta[log.ActivityID] = anchorMeta{} activityIDs = append(activityIDs, log.ActivityID) } if len(activityIDs) == 0 { return result } var rows []struct { ID int64 ChannelID int64 ChannelCode string } if err := s.repo.GetDbR().WithContext(ctx). Table("livestream_activities"). Select("id, channel_id, channel_code"). Where("id IN ?", activityIDs). Scan(&rows).Error; err != nil { s.logger.Error("[自动发放] 查询活动渠道信息失败", zap.Error(err)) return result } for _, row := range rows { activityMeta[row.ID] = anchorMeta{ channelID: row.ChannelID, channelCode: row.ChannelCode, } } missingChannelIDs := make([]int64, 0) seenChannels := make(map[int64]struct{}) for _, meta := range activityMeta { if meta.channelCode == "" && meta.channelID > 0 { if _, ok := seenChannels[meta.channelID]; !ok { seenChannels[meta.channelID] = struct{}{} missingChannelIDs = append(missingChannelIDs, meta.channelID) } } } channelCodeMap := s.fetchChannelCodes(ctx, missingChannelIDs) for activityID, meta := range activityMeta { code := meta.channelCode if code == "" && meta.channelID > 0 { code = channelCodeMap[meta.channelID] } if code != "" { result[activityID] = code } } return result } func (s *service) fetchChannelCodes(ctx context.Context, ids []int64) map[int64]string { result := make(map[int64]string) if len(ids) == 0 { return result } var rows []struct { ID int64 Code string } if err := s.repo.GetDbR().WithContext(ctx). Table("channels"). Select("id, code"). Where("id IN ?", ids). Scan(&rows).Error; err != nil { s.logger.Error("[自动发放] 查询渠道失败", zap.Error(err)) return result } for _, row := range rows { result[row.ID] = row.Code } return result } func (s *service) bindAnchorInviterIfNeeded(ctx context.Context, userID int64, anchorCode string) { if userID <= 0 || anchorCode == "" { return } userRecord, err := s.readDB.Users.WithContext(ctx). Select(s.readDB.Users.InviterID). Where(s.readDB.Users.ID.Eq(userID)). First() if err != nil { if !errors.Is(err, gorm.ErrRecordNotFound) { s.logger.Warn("[自动发放] 查询用户邀请人失败", zap.Int64("user_id", userID), zap.Error(err)) } return } if userRecord.InviterID != 0 { return } if _, err := s.userSvc.BindInviter(ctx, userID, user.BindInviterInput{InviteCode: anchorCode}); err != nil { if err == user.ErrAlreadyBound { return } if err == user.ErrInvalidCode { s.logger.Warn("[自动发放] 主播邀请码无效", zap.String("channel_code", anchorCode), zap.Int64("user_id", userID)) return } s.logger.Warn("[自动发放] 绑定主播邀请码失败", zap.String("channel_code", anchorCode), zap.Int64("user_id", userID), zap.Error(err)) return } s.logger.Info("[自动发放] 已补绑定主播邀请人", zap.Int64("user_id", userID), zap.String("channel_code", anchorCode)) }