package service import ( "context" "encoding/json" "errors" "fmt" "log" "math" "strconv" "strings" "time" "affiliate_dash/internal/model" "gorm.io/gorm" ) const ( rockPurchaseNoteAutoPrefix = "补货" rockPurchaseNoteManualPrefix = "手动采购" ) // ErrAutoPurchaseCooldown 表示自动补货仍在冷却窗口内,本次不应产生任何采购流水。 var ErrAutoPurchaseCooldown = errors.New("自动采购处于冷却中") // rockAutoPurchaseCooldownRemaining 返回自动补货还需等待的时长,0 表示可以执行。 // 冷却只认最近一次“自动”尝试:手动采购不影响自动补货节奏。 func rockAutoPurchaseCooldownRemaining(cfg *model.RockConfig, now time.Time) time.Duration { if cfg == nil || cfg.PurchaseCooldownMinutes <= 0 || cfg.LastAutoAttemptAt == nil { return 0 } cooldown := time.Duration(cfg.PurchaseCooldownMinutes) * time.Minute elapsed := now.Sub(*cfg.LastAutoAttemptAt) if elapsed >= cooldown { return 0 } return cooldown - elapsed } // PurchaseTokens 执行采购并入库,同时记录完整采购流水。 // 流程:积分/库存预检 → 降级计算 → claim 批量提取 → 创建上游打包记录 → 拉取包详情 → 幂等导入本地。 // 后续任一步失败都会用 claim 数据兜底入库,保证已扣积分的 Token 不会丢失。 func (s *RockService) PurchaseTokens(ctx context.Context, count int, isAuto bool) ([]model.RockToken, error) { s.mu.Lock() defer s.mu.Unlock() startedAt := time.Now() // 冷却判断必须在创建流水之前:被拦下的触发不留下任何“未采购”记录。 if isAuto { cfg, err := s.GetConfig(ctx) if err != nil { return nil, err } if remain := rockAutoPurchaseCooldownRemaining(cfg, startedAt); remain > 0 { return nil, fmt.Errorf("%w(冷却时间 %d 分钟)", ErrAutoPurchaseCooldown, cfg.PurchaseCooldownMinutes) } } trigger := "manual" if isAuto { trigger = "auto" } rec := newPurchaseRecorder(s.db, trigger, count, startedAt) tokens, err := s.purchaseTokensWithRecorder(ctx, count, isAuto, rec) s.lastSummary = rec.finish(tokens, err, time.Since(startedAt)) return tokens, err } // purchaseTokensWithRecorder 是采购主体,rec 负责记录各阶段的原始请求与响应。 func (s *RockService) purchaseTokensWithRecorder(ctx context.Context, count int, isAuto bool, rec *purchaseRecorder) ([]model.RockToken, error) { if count <= 0 { return nil, fmt.Errorf("采购数量必须大于 0") } cfg, err := s.GetConfig(ctx) if err != nil { return nil, err } if strings.TrimSpace(cfg.BearerToken) == "" { return nil, fmt.Errorf("尚未配置上游 Bearer Token,无法执行采购") } now := time.Now() // 记录本次尝试;自动补货同时推进自动冷却,手动采购只更新总尝试时间。 s.markPurchaseAttempt(ctx, cfg.ID, now, isAuto) // 把采购追踪器挂到 context,客户端会把每个上游调用的原始报文交给 rec。 ctx = WithRockTrace(ctx, &RockTrace{OnStage: rec.addStage}) // 采购前先取实时积分:积分决定这轮最多能买多少个。 points := cfg.Points rec.pointsSource = "snapshot" if me, meErr := s.client.GetMe(ctx, cfg.BearerToken); meErr == nil && me != nil { points = me.User.Points.Float64() rec.pointsSource = "live" } else if meErr != nil { log.Printf("[RockService] 获取实时积分失败,回退快照 %.2f: %v", points, meErr) } rec.pointsBefore = points // 再刷新 dashboard:拿到上游可用库存与最新单价(同时同步单价到配置)。 dashboard, dashErr := s.refreshDashboardWithConfig(ctx, cfg) if dashErr != nil { wrapped := fmt.Errorf("获取上游库存失败,已取消本次采购: %w", dashErr) s.recordPurchaseFailure(ctx, cfg.ID, wrapped.Error()) return nil, wrapped } rec.upstreamAvailable = dashboard.AvailableBlack unitCost := dashboard.BlackTokenCost if unitCost <= 0 { unitCost = cfg.BlackTokenCost } rec.unitCost = unitCost if unitCost <= 0 { err := errors.New("采购单价无效,无法计算可采购数量") s.recordPurchaseFailure(ctx, cfg.ID, err.Error()) return nil, err } maxByPoints := int(math.Floor(points / unitCost)) if maxByPoints < 0 { maxByPoints = 0 } maxByStock := dashboard.AvailableBlack if maxByStock < 0 { maxByStock = 0 } rec.maxByPoints = maxByPoints rec.maxByStock = maxByStock planned := count if maxByPoints < planned { planned = maxByPoints } if maxByStock < planned { planned = maxByStock } if planned < 0 { planned = 0 } rec.planned = planned rec.limitReason = purchaseLimitReason(count, planned, maxByPoints, maxByStock) if planned <= 0 { err := fmt.Errorf("本次未采购:积分 %.2f 点最多可买 %d 个,上游可用库存 %d 个", points, maxByPoints, maxByStock) s.recordPurchaseFailure(ctx, cfg.ID, err.Error()) return nil, err } if planned < count { log.Printf("[RockService] 采购数量降级:计划 %d 个,实际 %d 个(原因: %s)", count, planned, rec.limitReason) } claimed, err := s.client.ClaimTokens(ctx, cfg.BearerToken, model.RockColorBlack, planned) if err != nil { wrapped := fmt.Errorf("调用上游采购接口失败: %w", err) s.recordPurchaseFailure(ctx, cfg.ID, wrapped.Error()) return nil, wrapped } rec.claimed = len(claimed) if len(claimed) == 0 { err := errors.New("上游未返回任何有效 Token") s.recordPurchaseFailure(ctx, cfg.ID, err.Error()) return nil, err } // 积分已经扣除,立即刷新最后采购时间,后续打包失败也不会被重复采购。 s.markPurchased(ctx, cfg.ID, now) s.refreshPointsAsync(cfg.ID, cfg.BearerToken) ids := make([]string, 0, len(claimed)) claimedSet := make(map[string]struct{}, len(claimed)) for _, t := range claimed { if t.ID <= 0 { continue } upstreamID := strconv.FormatInt(t.ID, 10) ids = append(ids, upstreamID) claimedSet[upstreamID] = struct{}{} } if len(ids) == 0 { err := errors.New("上游返回的 Token ID 均无效") s.recordPurchaseFailure(ctx, cfg.ID, err.Error()) return nil, err } // 1. 创建上游打包记录 note := purchaseNote(isAuto, now) pkgResp, pkgErr := s.client.CreatePackage(ctx, cfg.BearerToken, ids, note) var packageID int64 shareCode := "" if pkgErr == nil && pkgResp != nil { packageID, shareCode = pkgResp.ResolvePackage() if shareCode == "" { pkgErr = errors.New("上游未返回 package.share_code") } } else if pkgErr == nil { pkgErr = errors.New("上游打包接口未返回内容") } if pkgErr != nil { reason := "创建上游打包记录失败: " + pkgErr.Error() rec.markDegraded(reason) return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, 0, "", "purchase", now, reason) } // 2. 拉取包详情(带重试) detail, detailErr := s.getPackageDetailWithRetry(ctx, shareCode) pid := packageID rec.packageID = &pid rec.shareCode = shareCode if detailErr != nil { reason := fmt.Sprintf("包 %s 详情拉取失败: %v", shareCode, detailErr) rec.markDegraded(reason) return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, shareCode, "package", now, reason) } // 3. 导入包内 Token:本批刚 claim 的 Token 强制 available,其余沿用上游状态 inserted, updated, skipped, importErr := s.importPackageDetail(ctx, detail, claimedSet) if importErr != nil { reason := fmt.Sprintf("包 %s 导入失败: %v", shareCode, importErr) rec.markDegraded(reason) return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, shareCode, "package", now, reason) } if inserted+updated == 0 { reason := fmt.Sprintf("包 %s 未返回有效 Token", shareCode) rec.markDegraded(reason) return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, shareCode, "package", now, reason) } // 4. 极端情况下详情缺 Token:用 claim 数据补齐,保证每个已扣积分的 Token 都在本地 ensureErr := s.ensureClaimedTokens(ctx, claimed, packageID, shareCode, "package", now) if ensureErr != nil { rec.markDegraded("补录缺失 Token 失败: " + ensureErr.Error()) s.recordPurchaseFailure(ctx, cfg.ID, "补录缺失 Token 失败: "+ensureErr.Error()) } dbTokens, findErr := s.findTokensByUpstreamIDs(ctx, ids) if findErr != nil || len(dbTokens) == 0 { if findErr != nil { log.Printf("[RockService] 回查采购 Token 失败: %v", findErr) } dbTokens = s.buildClaimedTokens(claimed, packageID, shareCode, "package", now) } rec.imported = len(dbTokens) if ensureErr == nil { s.recordPurchaseSuccess(ctx, cfg.ID) } log.Printf("[RockService] 采购打包完成: 包=%s 新增=%d 更新=%d 跳过=%d 实际入库=%d", shareCode, inserted, updated, skipped, len(dbTokens)) return dbTokens, nil } // fallbackClaimedTokens 在打包/详情/导入失败时,用 claim 返回的数据直接入库,避免积分白花。 func (s *RockService) fallbackClaimedTokens(ctx context.Context, cfgID uint, claimed []RockClaimedToken, packageID int64, shareCode, source string, at time.Time, reason string) ([]model.RockToken, error) { tokens, err := s.insertClaimedTokens(ctx, claimed, packageID, shareCode, source, at) if err != nil { s.recordPurchaseFailure(ctx, cfgID, reason+";兜底入库失败: "+err.Error()) return nil, fmt.Errorf("%s;兜底入库失败: %w", reason, err) } s.recordPurchaseFailure(ctx, cfgID, fmt.Sprintf("%s;已兜底入库 %d 个 Token", reason, len(tokens))) return tokens, nil } // insertClaimedTokens 把 claim 结果直接写入库存表。 func (s *RockService) insertClaimedTokens(ctx context.Context, claimed []RockClaimedToken, packageID int64, shareCode, source string, at time.Time) ([]model.RockToken, error) { tokens := s.buildClaimedTokens(claimed, packageID, shareCode, source, at) if len(tokens) == 0 { return nil, errors.New("没有可入库的 Token") } if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { return tx.Create(&tokens).Error }); err != nil { return nil, err } return tokens, nil } // ensureClaimedTokens 补录本地缺失的 claim Token(详情返回不完整时使用)。 func (s *RockService) ensureClaimedTokens(ctx context.Context, claimed []RockClaimedToken, packageID int64, shareCode, source string, at time.Time) error { tokens := s.buildClaimedTokens(claimed, packageID, shareCode, source, at) if len(tokens) == 0 { return nil } ids := make([]string, 0, len(tokens)) for _, t := range tokens { ids = append(ids, t.UpstreamTokenID) } var existing []string if err := s.db.WithContext(ctx).Unscoped().Model(&model.RockToken{}). Where("upstream_token_id IN ?", ids). Pluck("upstream_token_id", &existing).Error; err != nil { return err } seen := make(map[string]bool, len(existing)) for _, id := range existing { seen[id] = true } missing := make([]model.RockToken, 0, len(tokens)) for _, t := range tokens { if !seen[t.UpstreamTokenID] { missing = append(missing, t) } } if len(missing) == 0 { return nil } return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { return tx.Create(&missing).Error }) } func (s *RockService) buildClaimedTokens(claimed []RockClaimedToken, packageID int64, shareCode, source string, at time.Time) []model.RockToken { out := make([]model.RockToken, 0, len(claimed)) for _, t := range claimed { if t.ID <= 0 { continue } raw, err := json.Marshal(t) if err != nil { raw = []byte(`{}`) } rec := model.RockToken{ UpstreamTokenID: strconv.FormatInt(t.ID, 10), Token: strings.TrimSpace(t.Token), RedeemCode: strings.TrimSpace(t.RedeemCode), Color: defaultValue(strings.TrimSpace(t.Color), model.RockColorBlack), Region: defaultValue(strings.TrimSpace(t.Region), "Q/V可用"), Status: model.RockStatusAvailable, RawData: string(raw), Source: source, ClaimedAt: &at, } if packageID > 0 { pid := packageID rec.PackageID = &pid rec.PackageShareCode = strings.TrimSpace(shareCode) } out = append(out, rec) } return out } func (s *RockService) findTokensByUpstreamIDs(ctx context.Context, ids []string) ([]model.RockToken, error) { if len(ids) == 0 { return nil, nil } var tokens []model.RockToken err := s.db.WithContext(ctx). Where("upstream_token_id IN ?", ids). Order("id ASC"). Find(&tokens).Error return tokens, err } // getPackageDetailWithRetry 拉取包详情,最多 3 次(1s/2s 退避)。 func (s *RockService) getPackageDetailWithRetry(ctx context.Context, shareCode string) (*RockPackageDetailResponse, error) { var lastErr error for attempt := 1; attempt <= 3; attempt++ { detail, err := s.client.GetPackageDetail(ctx, shareCode) if err == nil && detail != nil && detail.Package.ID != 0 { return detail, nil } if err == nil { err = errors.New("上游未返回 package.id") } lastErr = err if attempt < 3 { select { case <-ctx.Done(): return nil, ctx.Err() case <-time.After(time.Duration(attempt) * time.Second): } } } return nil, lastErr } func purchaseNote(isAuto bool, at time.Time) string { prefix := rockPurchaseNoteManualPrefix if isAuto { prefix = rockPurchaseNoteAutoPrefix } return prefix + "-" + at.Format("20060102-150405") } func (s *RockService) markPurchaseAttempt(ctx context.Context, cfgID uint, at time.Time, isAuto bool) { updates := map[string]interface{}{"last_attempt_at": at} if isAuto { updates["last_auto_attempt_at"] = at } if err := s.db.WithContext(ctx).Model(&model.RockConfig{}). Where("id = ?", cfgID). Updates(updates).Error; err != nil { log.Printf("[RockService] 记录采购尝试时间失败: %v", err) } } func (s *RockService) markPurchased(ctx context.Context, cfgID uint, at time.Time) { if err := s.db.WithContext(ctx).Model(&model.RockConfig{}). Where("id = ?", cfgID). Update("last_purchased_at", at).Error; err != nil { log.Printf("[RockService] 记录最后采购时间失败: %v", err) } } func (s *RockService) recordPurchaseFailure(ctx context.Context, cfgID uint, msg string) { msg = truncatePurchaseError(msg) log.Printf("[RockService][ALERT] 补货异常: %s", msg) if err := s.db.WithContext(ctx).Model(&model.RockConfig{}). Where("id = ?", cfgID). Updates(map[string]interface{}{ "last_error": msg, "last_error_at": time.Now(), }).Error; err != nil { log.Printf("[RockService] 写入补货错误失败: %v", err) } } func (s *RockService) recordPurchaseSuccess(ctx context.Context, cfgID uint) { if err := s.db.WithContext(ctx).Model(&model.RockConfig{}). Where("id = ?", cfgID). Updates(map[string]interface{}{ "last_error": "", "last_error_at": nil, }).Error; err != nil { log.Printf("[RockService] 清理补货错误失败: %v", err) } } func (s *RockService) refreshPointsAsync(cfgID uint, bearerToken string) { go func() { bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() me, meErr := s.client.GetMe(bgCtx, bearerToken) if meErr == nil && me != nil { _ = s.db.Model(&model.RockConfig{}).Where("id = ?", cfgID).Update("points", me.User.Points.Float64()).Error } }() } func truncatePurchaseError(msg string) string { msg = strings.TrimSpace(msg) runes := []rune(msg) if len(runes) <= 500 { return msg } return string(runes[:500]) + "..." } // purchaseLimitReason 判断本次采购被哪个上限截断,用于流水与页面提示。 func purchaseLimitReason(requested, planned, maxByPoints, maxByStock int) string { if planned >= requested { return "" } switch { case maxByPoints == planned && maxByStock == planned: return "points_and_stock" case maxByPoints == planned: return "points" case maxByStock == planned: return "upstream_stock" default: return "" } } // CheckStockAndAutoPurchase 水位检查与自动采购触发 func (s *RockService) CheckStockAndAutoPurchase(ctx context.Context) { cfg, err := s.GetConfig(ctx) if err != nil { return } var count int64 if err := s.db.WithContext(ctx).Model(&model.RockToken{}). Where("status = ? AND color = ?", model.RockStatusAvailable, model.RockColorBlack). Count(&count).Error; err != nil { log.Printf("[RockService] 统计可用库存失败: %v", err) return } if count > int64(cfg.TokenWarningStock) { return } log.Printf("[RockService] 可用库存预警: 当前仅存 %d 个 (预警阈值 %d)", count, cfg.TokenWarningStock) if !cfg.AutoPurchaseEnabled { return } // 冷却期内静默跳过:不调用 PurchaseTokens,也不会产生“未采购”流水。 if remain := rockAutoPurchaseCooldownRemaining(cfg, time.Now()); remain > 0 { return } buyCount := cfg.AutoPurchaseCount if buyCount <= 0 { buyCount = 50 } log.Printf("[RockService] 触发自动补货,采购数量: %d", buyCount) tokens, err := s.PurchaseTokens(ctx, buyCount, true) if err != nil { // 并发触发时后到的请求会被 PurchaseTokens 拦下,这里不再重复记错误日志。 if errors.Is(err, ErrAutoPurchaseCooldown) { return } log.Printf("[RockService] 自动补货失败: %v", err) return } log.Printf("[RockService] 自动补货成功,实际入库 %d 个 Token", len(tokens)) } // ListTokens 分页查询 Token 列表 func (s *RockService) ListTokens(ctx context.Context, page, pageSize int, status, search string) ([]model.RockToken, int64, error) { if page <= 0 { page = 1 } if pageSize <= 0 { pageSize = 20 } db := applyRockTokenListFilters(s.db.WithContext(ctx).Model(&model.RockToken{}), status, search) var total int64 if err := db.Count(&total).Error; err != nil { return nil, 0, err } var tokens []model.RockToken // 按本地 ID 倒序:ID 自增,等价于按入库时间倒序;分页稳定,且与列表首列 ID 的降序一致 err := db.Order("id DESC"). Offset((page - 1) * pageSize). Limit(pageSize). Find(&tokens).Error if err != nil { return nil, 0, err } for i := range tokens { tokens[i].TokenCode = tokens[i].Token } // 批量反查关联订单的商户单号(创建订单时的 ClientOrderNo) orderNos := make([]string, 0, len(tokens)) for _, t := range tokens { if t.AssignedOrderNo != "" { orderNos = append(orderNos, t.AssignedOrderNo) } } if len(orderNos) > 0 { var orders []struct { OrderNo string `gorm:"column:order_no"` ClientOrderNo string `gorm:"column:client_order_no"` } if scanErr := s.db.WithContext(ctx).Table("fulfillment_orders"). Select("order_no, client_order_no"). Where("order_no IN ?", orderNos). Scan(&orders).Error; scanErr == nil { clientMap := make(map[string]string, len(orders)) for _, o := range orders { clientMap[o.OrderNo] = o.ClientOrderNo } for i := range tokens { if cNo, ok := clientMap[tokens[i].AssignedOrderNo]; ok && cNo != "" { tokens[i].ClientOrderNo = cNo } } } } return tokens, total, nil }