feat:一键刷新与自动刷新覆盖绑定错误/使用超时/绑定超时

2026-11-08 18:56:54 +08:00
parent 51dcb9de5a
commit 61d2aa47ea
9 changed files with 303 additions and 153 deletions
@@ -1,113 +0,0 @@
-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
-}
@@ -98,7 +98,7 @@ func TestListTokens_OverdueFilters(t *testing.T) {
 	}
 }

-func TestRefreshMismatchTokens_SkipsWhenAutoRefreshDisabled(t *testing.T) {
+func TestRefreshStaleTokens_SkipsWhenAutoRefreshDisabled(t *testing.T) {
 	db := newRockTestDB(t)
 	var calls int32
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -123,7 +123,7 @@ func TestRefreshMismatchTokens_SkipsWhenAutoRefreshDisabled(t *testing.T) {
 		t.Fatalf("调整 updated_at 失败: %v", err)
 	}

-	svc.RefreshMismatchTokens(context.Background())
+	svc.RefreshStaleTokens(context.Background())

 	if atomic.LoadInt32(&calls) != 0 {
 		t.Fatalf("自动刷新关闭时不应调用上游")
@@ -137,7 +137,7 @@ func TestRefreshMismatchTokens_SkipsWhenAutoRefreshDisabled(t *testing.T) {
 	}
 }

-func TestRefreshAllMismatchTokens_FixesMismatch(t *testing.T) {
+func TestRefreshAllStaleTokens_FixesMismatch(t *testing.T) {
 	db := newRockTestDB(t)
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "application/json")
@@ -168,7 +168,7 @@ func TestRefreshAllMismatchTokens_FixesMismatch(t *testing.T) {
 		t.Fatalf("调整 updated_at 失败: %v", err)
 	}

-	result, err := svc.RefreshAllMismatchTokens(context.Background())
+	result, err := svc.RefreshAllStaleTokens(context.Background())
 	if err != nil {
 		t.Fatalf("一键刷新失败: %v", err)
 	}
@@ -184,3 +184,129 @@ func TestRefreshAllMismatchTokens_FixesMismatch(t *testing.T) {
 		t.Fatalf("应修复为 bound 并清空 mismatch 标记，实际 %+v", saved)
 	}
 }
+
+// 手动一键刷新除了绑定错误，还要覆盖「使用超时」和「绑定超时」两类 Token。
+func TestRefreshAllStaleTokens_IncludesOverdueTokens(t *testing.T) {
+	db := newRockTestDB(t)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		switch r.URL.Path {
+		case "/api/redeem/CODE-USING-OLD/refresh":
+			_ = json.NewEncoder(w).Encode(map[string]interface{}{
+				"token": map[string]interface{}{
+					"status":           "bound",
+					"bound_account":    "新角色",
+					"bound_account_id": "10001",
+				},
+			})
+		case "/api/redeem/CODE-BOUND-OLD/refresh":
+			_ = json.NewEncoder(w).Encode(map[string]interface{}{
+				"token": map[string]interface{}{
+					"status":           "redeemed",
+					"bound_account":    "旧角色",
+					"bound_account_id": "10002",
+					"redeemed_at":      "2026-11-07 12:00:00",
+				},
+			})
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	}))
+	defer server.Close()
+
+	svc := NewRockService(db, NewRockClient(server.URL))
+	setRockTestBearer(t, svc, "bearer-test")
+
+	old := time.Now().AddDate(0, 0, -10)
+	usingToken := model.RockToken{
+		UpstreamTokenID: "stale-using",
+		RedeemCode:      "CODE-USING-OLD",
+		Status:          model.RockStatusUsing,
+		UsingAt:         &old,
+	}
+	boundToken := model.RockToken{
+		UpstreamTokenID: "stale-bound",
+		RedeemCode:      "CODE-BOUND-OLD",
+		Status:          model.RockStatusBound,
+		BoundAt:         &old,
+	}
+	for _, token := range []*model.RockToken{&usingToken, &boundToken} {
+		if err := db.Create(token).Error; err != nil {
+			t.Fatalf("创建 Token 失败: %v", err)
+		}
+		if err := db.Model(&model.RockToken{}).Where("id = ?", token.ID).
+			UpdateColumn("updated_at", time.Now().Add(-time.Hour)).Error; err != nil {
+			t.Fatalf("调整 updated_at 失败: %v", err)
+		}
+	}
+
+	result, err := svc.RefreshAllStaleTokens(context.Background())
+	if err != nil {
+		t.Fatalf("一键刷新失败: %v", err)
+	}
+	if result.Scanned != 2 || result.Fixed != 2 || result.Remaining != 0 || result.Failed != 0 {
+		t.Fatalf("一键刷新结果不符: %+v", result)
+	}
+
+	var savedUsing, savedBound model.RockToken
+	if err := db.First(&savedUsing, usingToken.ID).Error; err != nil {
+		t.Fatalf("读取使用超时 Token 失败: %v", err)
+	}
+	if savedUsing.Status != model.RockStatusBound {
+		t.Fatalf("使用超时 Token 应同步为 bound，实际 %s", savedUsing.Status)
+	}
+	if err := db.First(&savedBound, boundToken.ID).Error; err != nil {
+		t.Fatalf("读取绑定超时 Token 失败: %v", err)
+	}
+	if savedBound.Status != model.RockStatusRedeemed || savedBound.RedeemedAt == nil {
+		t.Fatalf("绑定超时 Token 应同步为 redeemed，实际 %+v", savedBound)
+	}
+}
+
+// 自动刷新与一键刷新同规则：开关开启后也要覆盖使用超时 Token。
+func TestRefreshStaleTokens_IncludesOverdueWhenEnabled(t *testing.T) {
+	db := newRockTestDB(t)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]interface{}{
+			"token": map[string]interface{}{
+				"status":           "bound",
+				"bound_account":    "自动刷新角色",
+				"bound_account_id": "20001",
+			},
+		})
+	}))
+	defer server.Close()
+
+	svc := NewRockService(db, NewRockClient(server.URL))
+	setRockTestBearer(t, svc, "bearer-test")
+	enabled := true
+	if _, err := svc.UpdateConfig(context.Background(), UpdateRockConfigReq{AutoRefreshMismatchEnabled: &enabled}); err != nil {
+		t.Fatalf("开启自动刷新失败: %v", err)
+	}
+
+	old := time.Now().AddDate(0, 0, -10)
+	token := model.RockToken{
+		UpstreamTokenID: "auto-stale-using",
+		RedeemCode:      "CODE-AUTO-OLD",
+		Status:          model.RockStatusUsing,
+		UsingAt:         &old,
+	}
+	if err := db.Create(&token).Error; err != nil {
+		t.Fatalf("创建 Token 失败: %v", err)
+	}
+	if err := db.Model(&model.RockToken{}).Where("id = ?", token.ID).
+		UpdateColumn("updated_at", time.Now().Add(-time.Hour)).Error; err != nil {
+		t.Fatalf("调整 updated_at 失败: %v", err)
+	}
+
+	svc.RefreshStaleTokens(context.Background())
+
+	var saved model.RockToken
+	if err := db.First(&saved, token.ID).Error; err != nil {
+		t.Fatalf("读取 Token 失败: %v", err)
+	}
+	if saved.Status != model.RockStatusBound {
+		t.Fatalf("自动刷新也应处理使用超时 Token，实际 %s", saved.Status)
+	}
+}
@@ -0,0 +1,128 @@
+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
+}