package service import ( "context" "errors" "log" "sync/atomic" "time" "affiliate_dash/internal/model" ) const ( // 每次巡检/一键刷新最多处理 20 条,避免手动请求超时。 rockStaleScanLimit = 20 // 同一 Token 的处理冷却:10 分钟内不重复调用上游。 rockStaleScanCooldown = 10 * time.Minute // 逐条刷新间隔,避免触发上游限流。 rockStaleScanGap = 1500 * time.Millisecond ) // errRockStaleScanBusy 表示已有巡检/一键刷新在跑,本次直接跳过。 var errRockStaleScanBusy = errors.New("刷新任务正在执行") var rockStaleScanRunning atomic.Bool // RockStaleRefreshResult 是异常 Token 刷新(巡检/一键)的汇总结果。 type RockStaleRefreshResult struct { Scanned int `json:"scanned"` // 本次扫描条数 Fixed int `json:"fixed"` // 已同步为其它状态(视为修复)的条数 Failed int `json:"failed"` // 调用上游失败的条数 Remaining int `json:"remaining"` // 刷新后状态仍未变化的条数 } // RefreshStaleTokens 定时巡检入口:受「自动刷新」开关控制,关闭时直接跳过。 // // 与手动「一键刷新」使用同一套扫描规则:绑定错误 + 使用超时 + 绑定超时。 func (s *RockService) RefreshStaleTokens(ctx context.Context) { cfg, err := s.GetConfig(ctx) if err != nil { log.Printf("[Rock] 异常 Token 巡检读取配置失败: %v", err) return } if !cfg.AutoRefreshMismatchEnabled { return } result, err := s.refreshStaleTokens(ctx) if err != nil { if errors.Is(err, errRockStaleScanBusy) { return } log.Printf("[Rock] 异常 Token 巡检失败: %v", err) return } if result.Scanned == 0 { return } log.Printf("[Rock] 异常 Token 巡检完成: 扫描=%d 修复=%d 仍异常=%d 失败=%d", result.Scanned, result.Fixed, result.Remaining, result.Failed) } // RefreshAllStaleTokens 手动「一键刷新」:绑定错误 + 使用超时 + 绑定超时, // 忽略自动开关,同步执行并返回结果。 func (s *RockService) RefreshAllStaleTokens(ctx context.Context) (*RockStaleRefreshResult, error) { return s.refreshStaleTokens(ctx) } // refreshStaleTokens 扫描异常 Token,逐条刷新上游状态。 // // 选取条件:绑定错误,或使用中/已绑定超过阈值(与列表筛选同口径); // 且 updated_at 早于 10 分钟前(同一 Token 冷却),按最久未处理优先,单次最多 20 条。 // 刷新失败或状态未变化时,RefreshTokenBinding 会写 updated_at,从而进入下一轮冷却。 func (s *RockService) refreshStaleTokens(ctx context.Context) (*RockStaleRefreshResult, error) { if !rockStaleScanRunning.CompareAndSwap(false, true) { return nil, errRockStaleScanBusy } defer rockStaleScanRunning.Store(false) cfg, err := s.GetConfig(ctx) if err != nil { return nil, err } warnDays, _, boundWarnDays := rockOverdueThresholds(cfg) now := time.Now() cutoff := now.Add(-rockStaleScanCooldown) query := s.db.WithContext(ctx).Model(&model.RockToken{}). Where("updated_at <= ?", cutoff). Where( "(status = ? OR (status = ? AND using_at IS NOT NULL AND using_at <= ?) OR (status = ? AND bound_at IS NOT NULL AND bound_at <= ?))", model.RockStatusMismatch, model.RockStatusUsing, rockOverdueCutoff(now, warnDays), model.RockStatusBound, rockOverdueCutoff(now, boundWarnDays), ) var tokens []model.RockToken if err := query.Order("updated_at ASC").Limit(rockStaleScanLimit).Find(&tokens).Error; err != nil { return nil, err } result := &RockStaleRefreshResult{Scanned: len(tokens)} for i, token := range tokens { if i > 0 { select { case <-ctx.Done(): return result, ctx.Err() case <-time.After(rockStaleScanGap): } } res, err := s.RefreshTokenBinding(ctx, token.ID) if err != nil { result.Failed++ // 失败也推进 updated_at,保证同一 Token 至少 10 分钟后才会再次尝试。 _ = s.db.WithContext(ctx).Model(&model.RockToken{}). Where("id = ?", token.ID). Update("updated_at", time.Now()).Error log.Printf("[Rock] 刷新异常 Token %d 失败: %v", token.ID, err) continue } if res.Status == token.Status { result.Remaining++ continue } result.Fixed++ } return result, nil }