114 lines
3.6 KiB
Go
114 lines
3.6 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"affiliate_dash/internal/model"
|
|
)
|
|
|
|
const (
|
|
// 每次巡检/一键刷新最多处理 20 条，避免手动请求超时。
|
|
rockMismatchScanLimit = 20
|
|
// 同一 Token 的处理冷却：10 分钟内不重复调用上游。
|
|
rockMismatchScanCooldown = 10 * time.Minute
|
|
// 逐条刷新间隔，避免触发上游限流。
|
|
rockMismatchScanGap = 1500 * time.Millisecond
|
|
)
|
|
|
|
// errRockMismatchScanBusy 表示已有巡检/一键刷新在跑，本次直接跳过。
|
|
var errRockMismatchScanBusy = errors.New("绑定错误刷新任务正在执行")
|
|
|
|
var rockMismatchScanRunning atomic.Bool
|
|
|
|
// RockMismatchRefreshResult 是绑定错误刷新（巡检/一键）的汇总结果。
|
|
type RockMismatchRefreshResult struct {
|
|
Scanned int `json:"scanned"` // 本次扫描条数
|
|
Fixed int `json:"fixed"` // 已同步为正常状态的条数
|
|
Failed int `json:"failed"` // 调用上游失败的条数
|
|
Remaining int `json:"remaining"` // 仍停留在绑定错误的条数
|
|
}
|
|
|
|
// RefreshMismatchTokens 定时巡检入口：受「自动刷新」开关控制，关闭时直接跳过。
|
|
func (s *RockService) RefreshMismatchTokens(ctx context.Context) {
|
|
cfg, err := s.GetConfig(ctx)
|
|
if err != nil {
|
|
log.Printf("[Rock] 绑定错误巡检读取配置失败: %v", err)
|
|
return
|
|
}
|
|
if !cfg.AutoRefreshMismatchEnabled {
|
|
return
|
|
}
|
|
result, err := s.refreshMismatchTokens(ctx)
|
|
if err != nil {
|
|
if errors.Is(err, errRockMismatchScanBusy) {
|
|
return
|
|
}
|
|
log.Printf("[Rock] 绑定错误巡检失败: %v", err)
|
|
return
|
|
}
|
|
if result.Scanned == 0 {
|
|
return
|
|
}
|
|
log.Printf("[Rock] 绑定错误巡检完成: 扫描=%d 修复=%d 仍异常=%d 失败=%d",
|
|
result.Scanned, result.Fixed, result.Remaining, result.Failed)
|
|
}
|
|
|
|
// RefreshAllMismatchTokens 手动「一键刷新绑定错误」：忽略自动开关，同步执行并返回结果。
|
|
func (s *RockService) RefreshAllMismatchTokens(ctx context.Context) (*RockMismatchRefreshResult, error) {
|
|
return s.refreshMismatchTokens(ctx)
|
|
}
|
|
|
|
// refreshMismatchTokens 扫描绑定错误 Token，逐条刷新上游状态。
|
|
//
|
|
// 选取条件：status=mismatch 且 updated_at 早于 10 分钟前，按最久未处理优先，单次最多 20 条。
|
|
// 刷新失败或上游仍未绑定时，RefreshTokenBinding 会写 updated_at，从而进入下一轮冷却。
|
|
func (s *RockService) refreshMismatchTokens(ctx context.Context) (*RockMismatchRefreshResult, error) {
|
|
if !rockMismatchScanRunning.CompareAndSwap(false, true) {
|
|
return nil, errRockMismatchScanBusy
|
|
}
|
|
defer rockMismatchScanRunning.Store(false)
|
|
|
|
cutoff := time.Now().Add(-rockMismatchScanCooldown)
|
|
var tokens []model.RockToken
|
|
if err := s.db.WithContext(ctx).
|
|
Where("status = ?", model.RockStatusMismatch).
|
|
Where("updated_at <= ?", cutoff).
|
|
Order("updated_at ASC").
|
|
Limit(rockMismatchScanLimit).
|
|
Find(&tokens).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
result := &RockMismatchRefreshResult{Scanned: len(tokens)}
|
|
for i, token := range tokens {
|
|
if i > 0 {
|
|
select {
|
|
case <-ctx.Done():
|
|
return result, ctx.Err()
|
|
case <-time.After(rockMismatchScanGap):
|
|
}
|
|
}
|
|
|
|
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 == model.RockStatusMismatch {
|
|
result.Remaining++
|
|
continue
|
|
}
|
|
result.Fixed++
|
|
}
|
|
return result, nil
|
|
}
|