package voiceroomredpacket import ( "context" "errors" "fmt" "log" "time" "chatapp3-golang/internal/config" "chatapp3-golang/internal/integration" "chatapp3-golang/internal/model" "gorm.io/gorm" "gorm.io/gorm/clause" ) // StartRefundWorker 启动过期红包退款扫描。 func (s *Service) StartRefundWorker(ctx context.Context) { interval := refundInterval(s.cfg) ticker := time.NewTicker(interval) go func() { defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: if _, err := s.ProcessExpired(ctx, refundBatchSize(s.cfg)); err != nil { log.Printf("voice room red packet refund worker failed: %v", err) } } } }() } // ProcessExpired 扫描并处理过期红包退款。 func (s *Service) ProcessExpired(ctx context.Context, limit int) (*RefundResult, error) { if limit <= 0 { limit = defaultRefundBatchSize } var rows []model.VoiceRoomRedPacket if err := s.db.WithContext(ctx). Where("(status = ? AND expire_time <= ?) OR status = ?", packetStatusActive, time.Now(), packetStatusRefundFailed). Order("expire_time asc"). Limit(limit). Find(&rows).Error; err != nil { return nil, err } result := &RefundResult{Scanned: len(rows)} for _, row := range rows { if err := s.refundOne(ctx, row.PacketNo, false); err != nil { result.Failed++ log.Printf("voice room red packet refund failed packetNo=%s err=%v", row.PacketNo, err) continue } result.Success++ } return result, nil } // RetryRefund 后台手动重试指定红包退款。 func (s *Service) RetryRefund(ctx context.Context, packetNo string) error { return s.refundOne(ctx, packetNo, true) } func (s *Service) refundOne(ctx context.Context, packetNo string, force bool) error { lockKey := fmt.Sprintf("voice_room:red_packet:refund_lock:%s", packetNo) lockOK, err := s.redis.SetNX(ctx, lockKey, "1", 30*time.Second).Result() if err != nil { return err } if !lockOK { return nil } defer s.redis.Del(context.Background(), lockKey) var packet model.VoiceRoomRedPacket refundAmount := int64(0) if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). Where("packet_no = ?", packetNo). First(&packet).Error if errors.Is(err, gorm.ErrRecordNotFound) { return nil } if err != nil { return err } if packet.Status == packetStatusRefunded || packet.Status == packetStatusFinished { return nil } if packet.Status != packetStatusActive && packet.Status != packetStatusRefundFailed { return nil } if packet.Status == packetStatusActive && time.Now().Before(packet.ExpireTime) && !force { return nil } now := time.Now() if packet.RefundAmount > 0 && packet.Status == packetStatusRefundFailed { refundAmount = packet.RefundAmount } else { var total struct { Amount int64 } if err := tx.Model(&model.VoiceRoomRedPacketSlice{}). Select("COALESCE(SUM(amount), 0) AS amount"). Where("packet_id = ? AND status = ?", packet.ID, sliceStatusUnclaimed). Scan(&total).Error; err != nil { return err } refundAmount = total.Amount if err := tx.Model(&model.VoiceRoomRedPacketSlice{}). Where("packet_id = ? AND status = ?", packet.ID, sliceStatusUnclaimed). Updates(map[string]any{"status": sliceStatusReleased, "update_time": now}).Error; err != nil { return err } } if refundAmount <= 0 { return tx.Model(&model.VoiceRoomRedPacket{}). Where("id = ?", packet.ID). Updates(map[string]any{ "status": packetStatusFinished, "refund_status": refundStatusNone, "refund_amount": 0, "update_time": now, }).Error } packet.RefundWalletEventID = walletEventID(walletOriginRefund, packet.SysOrigin, packet.PacketNo, 0) packet.RefundAmount = refundAmount return tx.Model(&model.VoiceRoomRedPacket{}). Where("id = ?", packet.ID). Updates(map[string]any{ "status": packetStatusExpired, "refund_status": refundStatusPending, "refund_wallet_event_id": packet.RefundWalletEventID, "refund_amount": refundAmount, "update_time": now, }).Error }); err != nil { return err } if packet.ID == 0 || refundAmount <= 0 { return nil } if err := s.wallet.ChangeGoldBalance(ctx, integration.GoldReceiptCommand{ ReceiptType: walletReceiptIncome, UserID: packet.SenderUserID, SysOrigin: packet.SysOrigin, EventID: packet.RefundWalletEventID, Remark: fmt.Sprintf("voice room red packet refund %s", packet.PacketNo), Amount: integration.NewPennyAmountPayloadFromDollar(refundAmount), CloseDelayAsset: false, OpUserType: "APP", CustomizeOrigin: walletOriginRefund, CustomizeOriginDesc: walletOriginRefundDesc, }); err != nil { _ = s.db.WithContext(context.Background()).Model(&model.VoiceRoomRedPacket{}). Where("id = ?", packet.ID). Updates(map[string]any{ "status": packetStatusRefundFailed, "refund_status": refundStatusFailed, "error_message": trimErrorMessage(err.Error()), "update_time": time.Now(), }).Error return err } return s.db.WithContext(ctx).Model(&model.VoiceRoomRedPacket{}). Where("id = ?", packet.ID). Updates(map[string]any{ "status": packetStatusRefunded, "refund_status": refundStatusSuccess, "error_message": "", "update_time": time.Now(), }).Error } func refundInterval(cfg config.Config) time.Duration { if cfg.VoiceRoomRedPacket.RefundScanIntervalSeconds <= 0 { return defaultRefundInterval } return time.Duration(cfg.VoiceRoomRedPacket.RefundScanIntervalSeconds) * time.Second } func refundBatchSize(cfg config.Config) int { if cfg.VoiceRoomRedPacket.RefundBatchSize <= 0 { return defaultRefundBatchSize } return cfg.VoiceRoomRedPacket.RefundBatchSize }