feat:自动补货/手动采购逻辑修改:批量获取-》打包-》获取Token列表

This commit is contained in:
2026-10-29 16:05:39 +08:00
parent 8198bd6feb
commit 8494fabc82
23 changed files with 3679 additions and 85 deletions
@@ -0,0 +1,25 @@
-- ============================================================================
-- 015_rock_packages_final.sql
-- Rock 打包 Token 字段 + 采购失败冷却/告警字段(合并原 015/016/017 的最终态)
-- ============================================================================
-- 原 016:最终态不再保留上游流转时间字段
DROP INDEX IF EXISTS idx_rock_tokens_upstream_updated_at;
ALTER TABLE rock_tokens DROP COLUMN IF EXISTS upstream_updated_at;
-- 原 015:打包 Token 字段
ALTER TABLE rock_tokens ADD COLUMN IF NOT EXISTS package_id BIGINT;
ALTER TABLE rock_tokens ADD COLUMN IF NOT EXISTS raw_data TEXT NOT NULL DEFAULT '';
ALTER TABLE rock_tokens ADD COLUMN IF NOT EXISTS source VARCHAR(32) NOT NULL DEFAULT 'purchase';
CREATE INDEX IF NOT EXISTS idx_rock_tokens_package_id ON rock_tokens (package_id);
-- 导入幂等键:上游 0666 编号唯一
CREATE UNIQUE INDEX IF NOT EXISTS idx_rock_tokens_upstream_token_id
ON rock_tokens (upstream_token_id)
WHERE upstream_token_id <> '';
-- 原 017:采购尝试/失败冷却/告警
ALTER TABLE rock_configs ADD COLUMN IF NOT EXISTS last_attempt_at TIMESTAMPTZ;
ALTER TABLE rock_configs ADD COLUMN IF NOT EXISTS last_error TEXT NOT NULL DEFAULT '';
ALTER TABLE rock_configs ADD COLUMN IF NOT EXISTS last_error_at TIMESTAMPTZ;
@@ -0,0 +1,7 @@
-- ============================================================================
-- 016_rock_black_token_cost.sql
-- Rock 上游黑色 Token 采购单价(页面打开时从 /api/user/dashboard 同步)
-- ============================================================================
ALTER TABLE rock_configs
ADD COLUMN IF NOT EXISTS black_token_cost NUMERIC(12,2) NOT NULL DEFAULT 20.00;
+10
View File
@@ -69,6 +69,16 @@ func (h *RockHandler) GetStats(c *gin.Context) {
response.OK(c, stats)
}
// RefreshDashboard 拉取上游 dashboard 并同步最新采购单价。
func (h *RockHandler) RefreshDashboard(c *gin.Context) {
result, err := h.rockSvc.RefreshDashboard(c.Request.Context())
if err != nil {
response.BadRequest(c, err.Error())
return
}
response.OK(c, result)
}
type purchaseReq struct {
Count int `json:"count"`
}
+8
View File
@@ -41,11 +41,15 @@ type RockConfig struct {
Username string `gorm:"size:128;not null;default:''" json:"username"`
Points float64 `gorm:"type:numeric(12,2);not null;default:0.00" json:"points"`
MinPointsWarning float64 `gorm:"type:numeric(12,2);not null;default:1050.00" json:"min_points_warning"`
BlackTokenCost float64 `gorm:"type:numeric(12,2);not null;default:20.00" json:"black_token_cost"`
TokenWarningStock int `gorm:"not null;default:10" json:"token_warning_stock"`
AutoPurchaseEnabled bool `gorm:"not null;default:false" json:"auto_purchase_enabled"`
AutoPurchaseCount int `gorm:"not null;default:50" json:"auto_purchase_count"`
PurchaseCooldownMinutes int `gorm:"not null;default:10" json:"purchase_cooldown_minutes"`
LastPurchasedAt *time.Time `json:"last_purchased_at"`
LastAttemptAt *time.Time `json:"last_attempt_at,omitempty"` // 最近一次采购尝试(成功/失败)时间,用于失败冷却
LastError string `gorm:"type:text;not null;default:''" json:"last_error"` // 最近一次补货/打包错误,成功后清空
LastErrorAt *time.Time `json:"last_error_at,omitempty"`
LastHeartbeatAt *time.Time `json:"last_heartbeat_at"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
@@ -64,8 +68,10 @@ type RockToken struct {
Color string `gorm:"size:32;not null;default:'black';index" json:"color"`
Region string `gorm:"size:64;not null;default:'Q/V可用'" json:"region"`
Status string `gorm:"size:32;not null;default:'available';index" json:"status"`
PackageID *int64 `gorm:"index" json:"package_id,omitempty"`
AssignedOrderNo string `gorm:"size:128;index" json:"assigned_order_no"`
ClientOrderNo string `gorm:"-" json:"client_order_no,omitempty"` // 关联的创建订单商户单号
TokenCode string `gorm:"-" json:"token,omitempty"` // 提取码,仅管理端响应展示,不持久化
ExpectedUID string `gorm:"size:64" json:"expected_uid"`
ActualRoleName string `gorm:"size:128" json:"actual_role_name"`
ActualRoleID string `gorm:"size:64" json:"actual_role_id"`
@@ -78,6 +84,8 @@ type RockToken struct {
MismatchAt *time.Time `gorm:"index" json:"mismatch_at"`
MismatchReason string `gorm:"size:255" json:"mismatch_reason"`
Remark string `gorm:"type:text;not null;default:''" json:"remark"`
RawData string `gorm:"type:text;not null;default:''" json:"raw_data,omitempty"`
Source string `gorm:"size:32;not null;default:'purchase';index" json:"source"`
ClaimedAt *time.Time `json:"claimed_at"`
BoundAt *time.Time `json:"bound_at"`
RedeemedAt *time.Time `json:"redeemed_at"`
+1
View File
@@ -287,6 +287,7 @@ func Setup(h *Handlers) *gin.Engine {
platform.PUT("/platform/rock/config", h.Rock.UpdateConfig)
platform.POST("/platform/rock/test-conn", h.Rock.TestConnection)
platform.GET("/platform/rock/stats", h.Rock.GetStats)
platform.POST("/platform/rock/dashboard/refresh", h.Rock.RefreshDashboard)
platform.POST("/platform/rock/purchase", h.Rock.PurchaseTokens)
platform.GET("/platform/rock/tokens", h.Rock.ListTokens)
platform.POST("/platform/rock/tokens/:id/expected-uid", h.Rock.CorrectTokenUID)
@@ -394,6 +394,10 @@ func TestRockService_StatsAndPurchase(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
if r.URL.Path == "/api/user/dashboard" {
writeRockDashboardJSON(w, 999, "20.0")
return
}
if r.URL.Path == "/api/me" {
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"user": map[string]interface{}{
+248
View File
@@ -249,6 +249,178 @@ type RockExchangeRecordsResponse struct {
Error string `json:"error,omitempty"`
}
// RockPackage 是上游打包记录(/api/user/packages)的列表项。
type RockPackage struct {
ID int64 `json:"id"`
UserID int64 `json:"user_id"`
TokenCount int `json:"token_count"`
ShareCode string `json:"share_code"`
Status string `json:"status"`
CreatedAt string `json:"created_at"`
Note string `json:"note"`
}
// RockPackageToken 是打包详情(/api/package/{share_code})里的 Token 原始结构。
// 字段命名与 0666.pro 返回保持一致,便于把整条 JSON 原样存入 raw_data。
type RockPackageToken struct {
ID int64 `json:"id"`
RedeemCode string `json:"redeem_code"`
Token string `json:"token"`
Color string `json:"color"`
Region string `json:"region"`
Phone string `json:"phone"`
Status string `json:"status"`
BoundAccount string `json:"bound_account"`
BoundAccountID string `json:"bound_account_id"`
PreviousBoundAccount string `json:"previous_bound_account"`
PreviousBoundAccountID string `json:"previous_bound_account_id"`
BindChangedAt string `json:"bind_changed_at"`
BindChangeType string `json:"bind_change_type"`
RedeemSkin string `json:"redeem_skin"`
RedeemCurrency string `json:"redeem_currency"`
RedeemedAt string `json:"redeemed_at"`
ClaimGroup string `json:"claim_group"`
ClaimRound int `json:"claim_round"`
ClaimNo int `json:"claim_no"`
ExpiresAt string `json:"expires_at"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
type RockPackageDetailResponse struct {
Package RockPackage `json:"package"`
Owner RockPackageOwner `json:"owner"`
Tokens []RockPackageToken `json:"tokens"`
Config map[string]any `json:"config"`
Settings map[string]any `json:"settings"`
Error string `json:"error,omitempty"`
}
type RockPackageOwner struct {
Username string `json:"username"`
Phone string `json:"phone"`
}
// RockPackageCreateResponse 是 POST /api/user/packages 的返回结构。
// 上游示例:{"package":{"id":1512,"token_count":2,"share_code":"1448fffa85bb3d4f913c",...}}
type RockPackageCreateResponse struct {
Package *RockPackage `json:"package"`
Packages []RockPackage `json:"packages"`
ShareCode string `json:"share_code"`
URL string `json:"url"`
Message string `json:"message,omitempty"`
Error string `json:"error,omitempty"`
}
// ResolvePackage 兼容 package / packages / share_code / url 四种返回形态,返回包 ID 与 share_code。
func (r *RockPackageCreateResponse) ResolvePackage() (int64, string) {
if r == nil {
return 0, ""
}
if r.Package != nil {
if code := strings.TrimSpace(r.Package.ShareCode); code != "" {
return r.Package.ID, code
}
}
for _, p := range r.Packages {
if code := strings.TrimSpace(p.ShareCode); code != "" {
return p.ID, code
}
}
if code := strings.TrimSpace(r.ShareCode); code != "" {
return 0, code
}
if rawURL := strings.TrimSpace(r.URL); rawURL != "" {
if idx := strings.LastIndex(rawURL, "/"); idx >= 0 && idx < len(rawURL)-1 {
return 0, strings.TrimSpace(rawURL[idx+1:])
}
}
return 0, ""
}
// CreatePackage 把一批已提取 Token 打包成公开分享包(需 Bearer Token)。
// 注意:上游未返回 share_code 时不会在这里报错,由调用方决定兜底策略。
func (c *RockClient) CreatePackage(ctx context.Context, bearerToken string, ids []string, note string) (*RockPackageCreateResponse, error) {
if len(ids) == 0 {
return nil, fmt.Errorf("打包 ids 不能为空")
}
reqURL := fmt.Sprintf("%s/api/user/packages", c.BaseURL())
payload := map[string]interface{}{
"ids": ids,
"note": note,
}
jsonBytes, _ := json.Marshal(payload)
req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, bytes.NewReader(jsonBytes))
if err != nil {
return nil, err
}
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", strings.TrimSpace(bearerToken)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json, text/plain, */*")
req.Header.Set("Origin", c.BaseURL())
req.Header.Set("Referer", c.BaseURL()+"/")
resp, err := c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("请求 0666 packages 创建接口失败: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("读取 packages 创建响应失败: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("上游 packages 创建接口返回错误码 %d: %s", resp.StatusCode, string(body))
}
var res RockPackageCreateResponse
if err := json.Unmarshal(body, &res); err != nil {
return nil, fmt.Errorf("解析 packages 创建响应失败: %w", err)
}
if res.Error != "" {
return nil, fmt.Errorf("上游 packages 创建业务错误: %s", res.Error)
}
return &res, nil
}
// GetPackageDetail 按 share_code 查询打包详情(公开接口,无需 Bearer)。
func (c *RockClient) GetPackageDetail(ctx context.Context, shareCode string) (*RockPackageDetailResponse, error) {
shareCode = strings.TrimSpace(shareCode)
if shareCode == "" {
return nil, fmt.Errorf("share_code 不能为空")
}
reqURL := fmt.Sprintf("%s/api/package/%s", c.BaseURL(), url.PathEscape(shareCode))
req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
if err != nil {
return nil, err
}
req.Header.Set("Accept", "application/json")
resp, err := c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("请求 0666 package 详情失败: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("读取 package 详情响应失败: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("上游 package 详情接口返回错误码 %d: %s", resp.StatusCode, string(body))
}
var res RockPackageDetailResponse
if err := json.Unmarshal(body, &res); err != nil {
return nil, fmt.Errorf("解析 package 详情失败: %w", err)
}
if res.Error != "" {
return nil, fmt.Errorf("上游 package 详情业务错误: %s", res.Error)
}
return &res, nil
}
// GetMe 获取商户个人信息及积分
func (c *RockClient) GetMe(ctx context.Context, bearerToken string) (*RockMeResponse, error) {
reqURL := fmt.Sprintf("%s/api/me", c.BaseURL())
@@ -287,6 +459,82 @@ func (c *RockClient) GetMe(ctx context.Context, bearerToken string) (*RockMeResp
return &res, nil
}
// RockDashboardResponse 是 GET /api/user/dashboard 的返回结构。
type RockDashboardResponse struct {
Available struct {
Black int `json:"black"`
Blue int `json:"blue"`
} `json:"available"`
EffectiveTokenCosts struct {
BlackTokenCost FlexibleFloat64 `json:"blackTokenCost"`
BlueTokenCost FlexibleFloat64 `json:"blueTokenCost"`
} `json:"effectiveTokenCosts"`
Config struct {
BlackTokenCost FlexibleFloat64 `json:"blackTokenCost"`
BlueTokenCost FlexibleFloat64 `json:"blueTokenCost"`
} `json:"config"`
Error string `json:"error,omitempty"`
}
// ResolveBlackTokenCost 优先取 effectiveTokenCosts,其次 config。
func (r *RockDashboardResponse) ResolveBlackTokenCost() float64 {
if r == nil {
return 0
}
if cost := float64(r.EffectiveTokenCosts.BlackTokenCost); cost > 0 {
return cost
}
return float64(r.Config.BlackTokenCost)
}
func (r *RockDashboardResponse) ResolveBlueTokenCost() float64 {
if r == nil {
return 0
}
if cost := float64(r.EffectiveTokenCosts.BlueTokenCost); cost > 0 {
return cost
}
return float64(r.Config.BlueTokenCost)
}
// GetDashboard 获取上游可用库存与实时采购单价(需 Bearer Token)。
func (c *RockClient) GetDashboard(ctx context.Context, bearerToken string) (*RockDashboardResponse, error) {
reqURL := fmt.Sprintf("%s/api/user/dashboard", c.BaseURL())
req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
if err != nil {
return nil, err
}
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", strings.TrimSpace(bearerToken)))
req.Header.Set("Accept", "application/json, text/plain, */*")
req.Header.Set("Origin", c.BaseURL())
req.Header.Set("Referer", c.BaseURL()+"/")
resp, err := c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("请求 0666 dashboard 失败: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("读取 dashboard 响应失败: %w", err)
}
if resp.StatusCode == http.StatusUnauthorized {
return nil, fmt.Errorf("上游 Bearer Token 已失效或未授权 (401)")
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("上游 dashboard 接口返回错误码 %d: %s", resp.StatusCode, string(body))
}
var res RockDashboardResponse
if err := json.Unmarshal(body, &res); err != nil {
return nil, fmt.Errorf("解析 dashboard 响应失败: %w", err)
}
if res.Error != "" {
return nil, fmt.Errorf("上游 dashboard 业务错误: %s", res.Error)
}
return &res, nil
}
// ClaimTokens 批量提取采购 Token
func (c *RockClient) ClaimTokens(ctx context.Context, bearerToken string, color string, count int) ([]RockClaimedToken, error) {
if count <= 0 {
@@ -0,0 +1,179 @@
package service
import (
"context"
"encoding/json"
"math"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
"affiliate_dash/internal/model"
)
// writeRockDashboardJSON 输出与 0666.pro 一致的 dashboard 结构,供各用例复用。
func writeRockDashboardJSON(w http.ResponseWriter, blackStock int, blackCost string) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"available": map[string]int{"black": blackStock, "blue": 7},
"effectiveTokenCosts": map[string]string{"blackTokenCost": blackCost, "blueTokenCost": "13.0"},
})
}
func TestRefreshDashboard_SameCostSkipsUpdate(t *testing.T) {
db := newRockTestDB(t)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/api/user/dashboard" {
w.WriteHeader(http.StatusNotFound)
return
}
if r.Header.Get("Authorization") != "Bearer bearer-test" {
w.WriteHeader(http.StatusUnauthorized)
return
}
writeRockDashboardJSON(w, 326, "20.0")
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
// 冻结 updated_at:价格未变化时不允许产生任何 UPDATE。
frozen := time.Date(2026, 10, 1, 8, 0, 0, 0, time.UTC)
if err := db.Model(&model.RockConfig{}).Where("id = 1").UpdateColumn("updated_at", frozen).Error; err != nil {
t.Fatalf("冻结 updated_at 失败: %v", err)
}
res, err := svc.RefreshDashboard(context.Background())
if err != nil {
t.Fatalf("刷新 dashboard 失败: %v", err)
}
if res.AvailableBlack != 326 {
t.Fatalf("上游可用库存应为 326,实际 %d", res.AvailableBlack)
}
if math.Abs(res.BlackTokenCost-20) > 1e-9 {
t.Fatalf("采购单价应为 20,实际 %v", res.BlackTokenCost)
}
if res.CostUpdated {
t.Fatalf("价格一致时不应标记为已更新")
}
cfg, _ := svc.GetConfig(context.Background())
if math.Abs(cfg.BlackTokenCost-20) > 1e-9 {
t.Fatalf("价格一致时库值不应变化,实际 %v", cfg.BlackTokenCost)
}
if !cfg.UpdatedAt.Equal(frozen) {
t.Fatalf("价格一致时不应写库,updated_at 被改为 %v", cfg.UpdatedAt)
}
}
func TestRefreshDashboard_UpdatesChangedCost(t *testing.T) {
db := newRockTestDB(t)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/api/user/dashboard" {
w.WriteHeader(http.StatusNotFound)
return
}
writeRockDashboardJSON(w, 88, "21.5")
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
res, err := svc.RefreshDashboard(context.Background())
if err != nil {
t.Fatalf("刷新 dashboard 失败: %v", err)
}
if !res.CostUpdated {
t.Fatalf("价格变化时应标记为已更新")
}
if math.Abs(res.BlackTokenCost-21.5) > 1e-9 {
t.Fatalf("采购单价应为 21.5,实际 %v", res.BlackTokenCost)
}
cfg, _ := svc.GetConfig(context.Background())
if math.Abs(cfg.BlackTokenCost-21.5) > 1e-9 {
t.Fatalf("采购单价应同步为 21.5,实际 %v", cfg.BlackTokenCost)
}
}
func TestPurchaseTokens_BlocksWhenUpstreamStockInsufficient(t *testing.T) {
db := newRockTestDB(t)
var claimCalls int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/api/user/dashboard":
writeRockDashboardJSON(w, 1, "20.0")
case "/api/user/tokens/claim":
atomic.AddInt32(&claimCalls, 1)
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"tokens": []map[string]interface{}{
{"id": 41001, "token": "token-1", "color": "black", "region": "Q/V可用", "status": "using", "redeem_code": "code-1"},
},
})
default:
w.WriteHeader(http.StatusNotFound)
}
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
_, err := svc.PurchaseTokens(context.Background(), 2, false)
if err == nil || !strings.Contains(err.Error(), "上游可用库存不足") {
t.Fatalf("库存不足时应阻止采购,实际 err=%v", err)
}
if calls := atomic.LoadInt32(&claimCalls); calls != 0 {
t.Fatalf("库存不足时不应调用上游 claim,实际调用 %d 次", calls)
}
var count int64
db.Model(&model.RockToken{}).Count(&count)
if count != 0 {
t.Fatalf("库存不足时不应入库任何 Token,实际 %d 条", count)
}
cfg, _ := svc.GetConfig(context.Background())
if !strings.Contains(cfg.LastError, "上游可用库存不足") {
t.Fatalf("last_error 应记录库存不足,实际 %q", cfg.LastError)
}
}
func TestPurchaseTokens_BlocksWhenDashboardFails(t *testing.T) {
db := newRockTestDB(t)
var claimCalls int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/api/user/dashboard":
w.WriteHeader(http.StatusInternalServerError)
_, _ = w.Write([]byte(`{"error":"dashboard down"}`))
case "/api/user/tokens/claim":
atomic.AddInt32(&claimCalls, 1)
_ = json.NewEncoder(w).Encode(map[string]interface{}{"tokens": []interface{}{}})
default:
w.WriteHeader(http.StatusNotFound)
}
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
_, err := svc.PurchaseTokens(context.Background(), 2, true)
if err == nil || !strings.Contains(err.Error(), "获取上游库存失败,已取消本次采购") {
t.Fatalf("dashboard 查询失败时应阻止采购,实际 err=%v", err)
}
if calls := atomic.LoadInt32(&claimCalls); calls != 0 {
t.Fatalf("dashboard 失败时不应调用上游 claim,实际调用 %d 次", calls)
}
cfg, _ := svc.GetConfig(context.Background())
if !strings.Contains(cfg.LastError, "获取上游库存失败,已取消本次采购") {
t.Fatalf("last_error 应记录 dashboard 失败,实际 %q", cfg.LastError)
}
}
+170
View File
@@ -0,0 +1,170 @@
package service
import (
"context"
"encoding/json"
"errors"
"strconv"
"strings"
"time"
"affiliate_dash/internal/model"
"gorm.io/gorm"
)
// importPackageDetail 把单个包详情幂等写入本地 Token,返回新增/更新/跳过数量。
func (s *RockService) importPackageDetail(ctx context.Context, detail *RockPackageDetailResponse, forceAvailableIDs map[string]struct{}) (int, int, int, error) {
if detail == nil || detail.Package.ID == 0 {
return 0, 0, 0, errors.New("上游未返回 package.id")
}
packageID := detail.Package.ID
inserted, updated, skipped := 0, 0, 0
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
for _, pt := range detail.Tokens {
if pt.ID <= 0 {
skipped++
continue
}
upstreamID := strconv.FormatInt(pt.ID, 10)
status := normalizeRockPackageStatus(pt.Status)
raw, marshalErr := json.Marshal(pt)
if marshalErr != nil {
raw = []byte(`{}`)
}
claimedAt := parseRockPackageTime(pt.CreatedAt)
boundAt := parseRockPackageTime(pt.BindChangedAt)
redeemedAt := parseRockPackageTime(pt.RedeemedAt)
var existing model.RockToken
findErr := tx.Where("upstream_token_id = ?", upstreamID).First(&existing).Error
if errors.Is(findErr, gorm.ErrRecordNotFound) {
insertStatus := status
if _, ok := forceAvailableIDs[upstreamID]; ok {
insertStatus = model.RockStatusAvailable
}
rec := model.RockToken{
UpstreamTokenID: upstreamID,
Token: strings.TrimSpace(pt.Token),
RedeemCode: strings.TrimSpace(pt.RedeemCode),
Color: defaultValue(strings.TrimSpace(pt.Color), model.RockColorBlack),
Region: defaultValue(strings.TrimSpace(pt.Region), "Q/V可用"),
Status: insertStatus,
PackageID: &packageID,
ActualRoleName: strings.TrimSpace(pt.BoundAccount),
ActualRoleID: strings.TrimSpace(pt.BoundAccountID),
ActualChannel: strings.TrimSpace(pt.Region),
TargetSkin: strings.TrimSpace(pt.RedeemSkin),
TargetCurrency: defaultValue(strings.TrimSpace(pt.RedeemCurrency), "幸运币"),
TargetCurrencyQuantity: 0,
RawData: string(raw),
Source: "package",
ClaimedAt: claimedAt,
BoundAt: boundAt,
RedeemedAt: redeemedAt,
}
if err := tx.Create(&rec).Error; err != nil {
return err
}
inserted++
continue
}
if findErr != nil {
return findErr
}
updates := map[string]interface{}{
"package_id": packageID,
"raw_data": string(raw),
}
// 已存在的本地 Token 只补空字段,避免覆盖手动采购/正在履约的本地状态。
if existing.Token == "" {
updates["token"] = strings.TrimSpace(pt.Token)
}
if existing.RedeemCode == "" {
updates["redeem_code"] = strings.TrimSpace(pt.RedeemCode)
}
if existing.Color == "" {
updates["color"] = defaultValue(strings.TrimSpace(pt.Color), model.RockColorBlack)
}
if existing.Region == "" {
updates["region"] = defaultValue(strings.TrimSpace(pt.Region), "Q/V可用")
}
if existing.Source == "" {
updates["source"] = "package"
}
// 打包导入且没有本地订单的记录:允许随上游刷新展示字段;
// 但手工释放为 available 的记录保持 available,避免全量导入把可售库存回退。
canRefreshUpstream := existing.Source == "package" &&
strings.TrimSpace(existing.AssignedOrderNo) == "" &&
existing.Status != model.RockStatusAvailable
if canRefreshUpstream {
updates["status"] = status
if strings.TrimSpace(pt.BoundAccount) != "" {
updates["actual_role_name"] = strings.TrimSpace(pt.BoundAccount)
}
if strings.TrimSpace(pt.BoundAccountID) != "" {
updates["actual_role_id"] = strings.TrimSpace(pt.BoundAccountID)
}
if strings.TrimSpace(pt.Region) != "" {
updates["actual_channel"] = strings.TrimSpace(pt.Region)
}
if strings.TrimSpace(pt.RedeemSkin) != "" {
updates["target_skin"] = strings.TrimSpace(pt.RedeemSkin)
}
if strings.TrimSpace(pt.RedeemCurrency) != "" {
updates["target_currency"] = strings.TrimSpace(pt.RedeemCurrency)
}
if claimedAt != nil {
updates["claimed_at"] = claimedAt
}
if boundAt != nil {
updates["bound_at"] = boundAt
}
if redeemedAt != nil {
updates["redeemed_at"] = redeemedAt
}
}
if err := tx.Model(&existing).Updates(updates).Error; err != nil {
return err
}
updated++
}
return nil
})
return inserted, updated, skipped, err
}
func normalizeRockPackageStatus(status string) string {
switch strings.TrimSpace(status) {
case model.RockStatusAvailable:
return model.RockStatusAvailable
case model.RockStatusUsing:
return model.RockStatusUsing
case model.RockStatusBound:
return model.RockStatusBound
case model.RockStatusRedeemed:
return model.RockStatusRedeemed
case model.RockStatusMismatch:
return model.RockStatusMismatch
default:
// 未知上游状态默认落 abnormal,避免被错误分配为可用库存。
return model.RockStatusAbnormal
}
}
func parseRockPackageTime(raw string) *time.Time {
if t, ok := parseRockTime(raw); ok {
return &t
}
return nil
}
func defaultValue(value, fallback string) string {
if strings.TrimSpace(value) == "" {
return fallback
}
return value
}
@@ -0,0 +1,272 @@
package service
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"affiliate_dash/internal/model"
)
func setRockTestBearer(t *testing.T, svc *RockService, token string) {
t.Helper()
bearer := token
if _, err := svc.UpdateConfig(context.Background(), UpdateRockConfigReq{BearerToken: &bearer}); err != nil {
t.Fatalf("设置 Bearer Token 失败: %v", err)
}
}
func TestPurchaseTokens_CreatesPackageAndImportsAvailable(t *testing.T) {
db := newRockTestDB(t)
var gotClaimCount string
var gotClaimColor string
var gotIDs []string
var gotNote string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
switch {
case r.URL.Path == "/api/user/dashboard":
writeRockDashboardJSON(w, 999, "20.0")
case r.URL.Path == "/api/user/tokens/claim":
var body struct {
Color string `json:"color"`
Count string `json:"count"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
gotClaimColor = body.Color
gotClaimCount = body.Count
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"tokens": []map[string]interface{}{
{"id": 30001, "token": "token-1", "color": "black", "region": "Q/V可用", "status": "using", "redeem_code": "code-1"},
{"id": 30002, "token": "token-2", "color": "black", "region": "Q/V可用", "status": "using", "redeem_code": "code-2"},
},
})
case r.URL.Path == "/api/user/packages" && r.Method == http.MethodPost:
var body struct {
IDs []string `json:"ids"`
Note string `json:"note"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
gotIDs = body.IDs
gotNote = body.Note
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"package": map[string]interface{}{
"id": 9001, "token_count": 2, "share_code": "share123", "status": "active", "note": body.Note,
},
})
case r.URL.Path == "/api/package/share123":
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"package": map[string]interface{}{"id": 9001, "token_count": 2, "share_code": "share123", "status": "active"},
"tokens": []map[string]interface{}{
{"id": 30001, "token": "token-1", "color": "black", "region": "Q/V可用", "status": "using", "redeem_code": "code-1", "created_at": "2026-10-29 13:45:27"},
{"id": 30002, "token": "token-2", "color": "black", "region": "Q/V可用", "status": "using", "redeem_code": "code-2", "created_at": "2026-10-29 13:45:27"},
},
})
case r.URL.Path == "/api/me":
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"user": map[string]interface{}{"username": "tester", "points": "880.0"},
})
default:
w.WriteHeader(http.StatusNotFound)
}
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
tokens, err := svc.PurchaseTokens(context.Background(), 2, true)
if err != nil {
t.Fatalf("采购失败: %v", err)
}
if len(tokens) != 2 {
t.Fatalf("期望返回 2 个 Token,实际 %d", len(tokens))
}
if gotClaimColor != model.RockColorBlack || gotClaimCount != "2" {
t.Fatalf("claim 参数不符: color=%s count=%s", gotClaimColor, gotClaimCount)
}
if len(gotIDs) != 2 || gotIDs[0] != "30001" || gotIDs[1] != "30002" {
t.Fatalf("打包 ids 不符: %v", gotIDs)
}
if !strings.HasPrefix(gotNote, "补货-") {
t.Fatalf("自动补货 note 前缀不符: %s", gotNote)
}
var rows []model.RockToken
if err := db.Order("id ASC").Find(&rows).Error; err != nil {
t.Fatalf("查询 Token 失败: %v", err)
}
if len(rows) != 2 {
t.Fatalf("期望本地 2 条 Token,实际 %d", len(rows))
}
for _, row := range rows {
if row.Status != model.RockStatusAvailable {
t.Fatalf("新 Token 状态应为 available,实际 %s", row.Status)
}
if row.Source != "package" {
t.Fatalf("新 Token 来源应为 package,实际 %s", row.Source)
}
if row.PackageID == nil || *row.PackageID != 9001 {
t.Fatalf("package_id 未写入: %+v", row.PackageID)
}
if row.Token == "" || row.RedeemCode == "" {
t.Fatalf("提取码/兑换码未写入: %+v", row)
}
}
cfg, _ := svc.GetConfig(context.Background())
if cfg.LastError != "" {
t.Fatalf("成功后应清空 last_error,实际 %q", cfg.LastError)
}
if cfg.LastPurchasedAt == nil || cfg.LastAttemptAt == nil {
t.Fatalf("last_purchased_at/last_attempt_at 未记录")
}
}
func TestPurchaseTokens_PackageFailureFallsBackToPurchase(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/user/dashboard":
writeRockDashboardJSON(w, 999, "20.0")
case "/api/user/tokens/claim":
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"tokens": []map[string]interface{}{
{"id": 31001, "token": "token-1", "color": "black", "region": "Q/V可用", "status": "using", "redeem_code": "code-1"},
},
})
case "/api/user/packages":
w.WriteHeader(http.StatusInternalServerError)
_, _ = w.Write([]byte(`{"error":"package down"}`))
default:
w.WriteHeader(http.StatusNotFound)
}
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
tokens, err := svc.PurchaseTokens(context.Background(), 1, false)
if err != nil {
t.Fatalf("打包失败时应兜底成功,实际报错: %v", err)
}
if len(tokens) != 1 {
t.Fatalf("期望兜底入库 1 个,实际 %d", len(tokens))
}
var row model.RockToken
if err := db.First(&row).Error; err != nil {
t.Fatalf("查询 Token 失败: %v", err)
}
if row.Status != model.RockStatusAvailable || row.Source != "purchase" || row.PackageID != nil {
t.Fatalf("兜底入库字段不符: status=%s source=%s package_id=%v", row.Status, row.Source, row.PackageID)
}
cfg, _ := svc.GetConfig(context.Background())
if !strings.Contains(cfg.LastError, "创建上游打包记录失败") {
t.Fatalf("last_error 应记录打包失败,实际 %q", cfg.LastError)
}
}
func TestPurchaseTokens_PackageDetailFailureFallsBackWithPackageID(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/user/dashboard":
writeRockDashboardJSON(w, 999, "20.0")
case "/api/user/tokens/claim":
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"tokens": []map[string]interface{}{
{"id": 32001, "token": "token-1", "color": "black", "region": "Q/V可用", "status": "using", "redeem_code": "code-1"},
},
})
case "/api/user/packages":
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"package": map[string]interface{}{"id": 9002, "token_count": 1, "share_code": "detailfail", "status": "active"},
})
case "/api/package/detailfail":
w.WriteHeader(http.StatusInternalServerError)
_, _ = w.Write([]byte(`{"error":"detail down"}`))
default:
w.WriteHeader(http.StatusNotFound)
}
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
tokens, err := svc.PurchaseTokens(context.Background(), 1, false)
if err != nil {
t.Fatalf("详情失败时应兜底成功,实际报错: %v", err)
}
if len(tokens) != 1 {
t.Fatalf("期望兜底入库 1 个,实际 %d", len(tokens))
}
var row model.RockToken
if err := db.First(&row).Error; err != nil {
t.Fatalf("查询 Token 失败: %v", err)
}
if row.Status != model.RockStatusAvailable || row.Source != "package" {
t.Fatalf("兜底入库字段不符: status=%s source=%s", row.Status, row.Source)
}
if row.PackageID == nil || *row.PackageID != 9002 {
t.Fatalf("兜底入库应保留 package_id=9002,实际 %v", row.PackageID)
}
cfg, _ := svc.GetConfig(context.Background())
if !strings.Contains(cfg.LastError, "详情拉取失败") {
t.Fatalf("last_error 应记录详情失败,实际 %q", cfg.LastError)
}
}
func TestAutoPurchase_FailureCooldown(t *testing.T) {
db := newRockTestDB(t)
var claimCalls int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
if r.URL.Path == "/api/user/dashboard" {
writeRockDashboardJSON(w, 999, "20.0")
return
}
if r.URL.Path == "/api/user/tokens/claim" {
atomic.AddInt32(&claimCalls, 1)
w.WriteHeader(http.StatusInternalServerError)
_, _ = w.Write([]byte(`{"error":"insufficient points"}`))
return
}
w.WriteHeader(http.StatusNotFound)
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
if _, err := svc.PurchaseTokens(context.Background(), 50, true); err == nil {
t.Fatalf("第一次补货应因上游 500 失败")
}
if _, err := svc.PurchaseTokens(context.Background(), 50, true); err == nil || !strings.Contains(err.Error(), "冷却中") {
t.Fatalf("失败后第二次应立即命中冷却,实际 err=%v", err)
}
if calls := atomic.LoadInt32(&claimCalls); calls != 1 {
t.Fatalf("冷却期内不应再次请求上游 claim,实际调用 %d 次", calls)
}
cfg, _ := svc.GetConfig(context.Background())
if cfg.LastAttemptAt == nil {
t.Fatalf("last_attempt_at 未记录")
}
if !strings.Contains(cfg.LastError, "调用上游采购接口失败") {
t.Fatalf("last_error 未记录采购失败: %q", cfg.LastError)
}
}
+63
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"math"
"strings"
"sync"
"time"
@@ -42,6 +43,16 @@ type RockStatsResponse struct {
AutoPurchaseEnabled bool `json:"auto_purchase_enabled"`
Username string `json:"username"`
LastHeartbeatAt string `json:"last_heartbeat_at,omitempty"`
LastPurchaseError string `json:"last_purchase_error,omitempty"`
LastPurchaseErrorAt string `json:"last_purchase_error_at,omitempty"`
}
type RockDashboardResult struct {
AvailableBlack int `json:"available_black"`
AvailableBlue int `json:"available_blue"`
BlackTokenCost float64 `json:"black_token_cost"`
BlueTokenCost float64 `json:"blue_token_cost"`
CostUpdated bool `json:"cost_updated"`
}
type UpdateRockConfigReq struct {
@@ -66,6 +77,7 @@ func (s *RockService) GetConfig(ctx context.Context) (*model.RockConfig, error)
BaseURL: model.RockDefaultBaseURL,
BearerToken: "",
MinPointsWarning: 1050.00,
BlackTokenCost: 20.00,
TokenWarningStock: 10,
AutoPurchaseEnabled: false,
AutoPurchaseCount: 50,
@@ -170,6 +182,53 @@ func (s *RockService) TestConnection(ctx context.Context, bearerToken string) (*
return me, nil
}
// RefreshDashboard 拉取上游 dashboard,并在采购单价变化时同步到本系统。
func (s *RockService) RefreshDashboard(ctx context.Context) (*RockDashboardResult, error) {
cfg, err := s.GetConfig(ctx)
if err != nil {
return nil, err
}
return s.refreshDashboardWithConfig(ctx, cfg)
}
// refreshDashboardWithConfig 复用已读取的配置,避免同一次采购内重复查库。
func (s *RockService) refreshDashboardWithConfig(ctx context.Context, cfg *model.RockConfig) (*RockDashboardResult, error) {
if cfg == nil {
return nil, errors.New("Rock 配置不存在")
}
if strings.TrimSpace(cfg.BearerToken) == "" {
return nil, errors.New("请先配置 0666.pro Bearer Token")
}
dash, err := s.client.GetDashboard(ctx, cfg.BearerToken)
if err != nil {
return nil, err
}
result := &RockDashboardResult{
AvailableBlack: dash.Available.Black,
AvailableBlue: dash.Available.Blue,
BlueTokenCost: dash.ResolveBlueTokenCost(),
}
cost := dash.ResolveBlackTokenCost()
if cost <= 0 {
result.BlackTokenCost = cfg.BlackTokenCost
return result, nil
}
result.BlackTokenCost = cost
// 价格一样就完全不写库;只有变化超过 0.001 才更新。
if math.Abs(cfg.BlackTokenCost-cost) > 0.001 {
if err := s.db.WithContext(ctx).Model(&model.RockConfig{}).
Where("id = ?", cfg.ID).
Update("black_token_cost", cost).Error; err != nil {
return nil, err
}
result.CostUpdated = true
}
return result, nil
}
// GetStats 获取概览统计
func (s *RockService) GetStats(ctx context.Context) (*RockStatsResponse, error) {
cfg, err := s.GetConfig(ctx)
@@ -186,6 +245,10 @@ func (s *RockService) GetStats(ctx context.Context) (*RockStatsResponse, error)
if cfg.LastHeartbeatAt != nil {
stats.LastHeartbeatAt = cfg.LastHeartbeatAt.Format("2006-01-02 15:04:05")
}
stats.LastPurchaseError = cfg.LastError
if cfg.LastErrorAt != nil {
stats.LastPurchaseErrorAt = cfg.LastErrorAt.Format("2006-01-02 15:04:05")
}
type CountResult struct {
Status string
+336 -57
View File
@@ -2,8 +2,11 @@ package service
import (
"context"
"encoding/json"
"errors"
"fmt"
"log"
"strconv"
"strings"
"time"
@@ -11,75 +14,340 @@ import (
"gorm.io/gorm"
)
// PurchaseTokens 执行采购并批量入库(通用黑色 Token)
const (
rockPurchaseNoteAutoPrefix = "补货"
rockPurchaseNoteManualPrefix = "手动采购"
)
// PurchaseTokens 执行采购并入库。
// 流程:claim 批量提取 → 创建上游打包记录 → 拉取包详情 → 幂等导入本地(新记录 available)。
// 后续任一步失败都会用 claim 数据兜底入库,保证已扣积分的 Token 不会丢失。
func (s *RockService) PurchaseTokens(ctx context.Context, count int, isAuto bool) ([]model.RockToken, error) {
s.mu.Lock()
defer s.mu.Unlock()
if count <= 0 {
return nil, fmt.Errorf("采购数量必须大于 0")
}
cfg, err := s.GetConfig(ctx)
if err != nil {
return nil, err
}
if cfg.BearerToken == "" {
if strings.TrimSpace(cfg.BearerToken) == "" {
return nil, fmt.Errorf("尚未配置上游 Bearer Token,无法执行采购")
}
// 自动补货防抖冷却检查
now := time.Now()
if isAuto && cfg.LastPurchasedAt != nil {
cooldown := time.Duration(cfg.PurchaseCooldownMinutes) * time.Minute
if now.Sub(*cfg.LastPurchasedAt) < cooldown {
return nil, fmt.Errorf("自动采购处于冷却中(冷却时间 %d 分钟)", cfg.PurchaseCooldownMinutes)
if isAuto {
// 冷却以最近一次“尝试”为准:失败也生效,避免上游故障时每个触发点反复请求。
last := cfg.LastAttemptAt
if last == nil {
last = cfg.LastPurchasedAt
}
if last != nil && cfg.PurchaseCooldownMinutes > 0 {
cooldown := time.Duration(cfg.PurchaseCooldownMinutes) * time.Minute
if now.Sub(*last) < cooldown {
return nil, fmt.Errorf("自动采购处于冷却中(冷却时间 %d 分钟)", cfg.PurchaseCooldownMinutes)
}
}
}
s.updatePurchaseAttempt(ctx, cfg.ID, now)
// 调上游 Claim 接口(统一下单黑色通用 Token)
tokens, err := s.client.ClaimTokens(ctx, cfg.BearerToken, model.RockColorBlack, count)
if err != nil {
return nil, fmt.Errorf("调用上游采购接口失败: %w", err)
// 采购前先刷新上游 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
}
if dashboard.AvailableBlack < count {
err := fmt.Errorf("上游可用库存不足:当前 %d 个,本次需要 %d 个", dashboard.AvailableBlack, count)
s.recordPurchaseFailure(ctx, cfg.ID, err.Error())
return nil, err
}
claimed, err := s.client.ClaimTokens(ctx, cfg.BearerToken, model.RockColorBlack, count)
if err != nil {
wrapped := fmt.Errorf("调用上游采购接口失败: %w", err)
s.recordPurchaseFailure(ctx, cfg.ID, wrapped.Error())
return nil, wrapped
}
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 {
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, 0, "purchase", now,
"创建上游打包记录失败: "+pkgErr.Error())
}
// 2. 拉取包详情(带重试)
detail, detailErr := s.getPackageDetailWithRetry(ctx, shareCode)
if detailErr != nil {
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, "package", now,
fmt.Sprintf("包 %s 详情拉取失败: %v", shareCode, detailErr))
}
// 3. 导入包内 Token:本批刚 claim 的 Token 强制 available,其余沿用上游状态
inserted, updated, skipped, importErr := s.importPackageDetail(ctx, detail, claimedSet)
if importErr != nil {
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, "package", now,
fmt.Sprintf("包 %s 导入失败: %v", shareCode, importErr))
}
if inserted+updated == 0 {
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, "package", now,
fmt.Sprintf("包 %s 未返回有效 Token", shareCode))
}
// 4. 极端情况下详情缺 Token:用 claim 数据补齐,保证每个已扣积分的 Token 都在本地
ensureErr := s.ensureClaimedTokens(ctx, claimed, packageID, "package", now)
if ensureErr != nil {
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, "package", now)
}
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, source string, at time.Time, reason string) ([]model.RockToken, error) {
tokens, err := s.insertClaimedTokens(ctx, claimed, packageID, 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, source string, at time.Time) ([]model.RockToken, error) {
tokens := s.buildClaimedTokens(claimed, packageID, source, at)
if len(tokens) == 0 {
return nil, fmt.Errorf("上游未返回任何有效 Token")
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
}
var dbTokens []model.RockToken
// ensureClaimedTokens 补录本地缺失的 claim Token(详情返回不完整时使用)。
func (s *RockService) ensureClaimedTokens(ctx context.Context, claimed []RockClaimedToken, packageID int64, source string, at time.Time) error {
tokens := s.buildClaimedTokens(claimed, packageID, source, at)
if len(tokens) == 0 {
return nil
}
ids := make([]string, 0, len(tokens))
for _, t := range tokens {
dbTokens = append(dbTokens, model.RockToken{
UpstreamTokenID: fmt.Sprintf("%d", t.ID),
Token: t.Token,
RedeemCode: t.RedeemCode,
Color: model.RockColorBlack,
Region: t.Region,
Status: model.RockStatusAvailable,
ClaimedAt: &now,
})
ids = append(ids, t.UpstreamTokenID)
}
// 事务写入库存表
err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Create(&dbTokens).Error; err != nil {
return err
var existing []string
if err := s.db.WithContext(ctx).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)
}
// 更新最后采购时间
return tx.Model(&model.RockConfig{}).Where("id = ?", cfg.ID).Update("last_purchased_at", now).Error
})
if err != nil {
return nil, fmt.Errorf("入库采购 Token 失败: %w", err)
}
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, 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
}
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) updatePurchaseAttempt(ctx context.Context, cfgID uint, at time.Time) {
if err := s.db.WithContext(ctx).Model(&model.RockConfig{}).
Where("id = ?", cfgID).
Update("last_attempt_at", at).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, cfg.BearerToken)
me, meErr := s.client.GetMe(bgCtx, bearerToken)
if meErr == nil && me != nil {
_ = s.db.Model(&model.RockConfig{}).Where("id = 1").Update("points", me.User.Points.Float64()).Error
_ = s.db.Model(&model.RockConfig{}).Where("id = ?", cfgID).Update("points", me.User.Points.Float64()).Error
}
}()
}
return dbTokens, nil
func truncatePurchaseError(msg string) string {
msg = strings.TrimSpace(msg)
runes := []rune(msg)
if len(runes) <= 500 {
return msg
}
return string(runes[:500]) + "..."
}
// CheckStockAndAutoPurchase 水位检查与自动采购触发
@@ -90,26 +358,32 @@ func (s *RockService) CheckStockAndAutoPurchase(ctx context.Context) {
}
var count int64
s.db.WithContext(ctx).Model(&model.RockToken{}).
if err := s.db.WithContext(ctx).Model(&model.RockToken{}).
Where("status = ? AND color = ?", model.RockStatusAvailable, model.RockColorBlack).
Count(&count)
if count <= int64(cfg.TokenWarningStock) {
log.Printf("[RockService] 可用库存预警: 当前仅存 %d 个 (预警阈值 %d)", count, cfg.TokenWarningStock)
if cfg.AutoPurchaseEnabled {
buyCount := cfg.AutoPurchaseCount
if buyCount <= 0 {
buyCount = 50
}
log.Printf("[RockService] 触发自动补货,采购数量: %d", buyCount)
_, err := s.PurchaseTokens(ctx, buyCount, true)
if err != nil {
log.Printf("[RockService] 自动补货失败: %v", err)
} else {
log.Printf("[RockService] 自动补货成功,新增 %d 个 Token", buyCount)
}
}
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
}
buyCount := cfg.AutoPurchaseCount
if buyCount <= 0 {
buyCount = 50
}
log.Printf("[RockService] 触发自动补货,采购数量: %d", buyCount)
tokens, err := s.PurchaseTokens(ctx, buyCount, true)
if err != nil {
log.Printf("[RockService] 自动补货失败: %v", err)
return
}
log.Printf("[RockService] 自动补货成功,实际入库 %d 个 Token", len(tokens))
}
// ListTokens 分页查询 Token 列表
@@ -127,8 +401,8 @@ func (s *RockService) ListTokens(ctx context.Context, page, pageSize int, status
}
if strings.TrimSpace(search) != "" {
term := "%" + strings.TrimSpace(search) + "%"
db = db.Where("CAST(id AS TEXT) = ? OR upstream_token_id = ? OR upstream_token_id LIKE ? OR assigned_order_no LIKE ? OR expected_uid LIKE ? OR actual_role_name LIKE ? OR actual_role_id LIKE ? OR redeem_code LIKE ? OR remark LIKE ? OR assigned_order_no IN (SELECT order_no FROM fulfillment_orders WHERE client_order_no LIKE ?)",
strings.TrimSpace(search), strings.TrimSpace(search), term, term, term, term, term, term, term, term)
db = db.Where("CAST(id AS TEXT) = ? OR upstream_token_id = ? OR upstream_token_id LIKE ? OR assigned_order_no LIKE ? OR expected_uid LIKE ? OR actual_role_name LIKE ? OR actual_role_id LIKE ? OR redeem_code LIKE ? OR remark LIKE ? OR CAST(package_id AS TEXT) LIKE ? OR token LIKE ? OR assigned_order_no IN (SELECT order_no FROM fulfillment_orders WHERE client_order_no LIKE ?)",
strings.TrimSpace(search), strings.TrimSpace(search), term, term, term, term, term, term, term, term, term, term)
}
var total int64
@@ -137,13 +411,18 @@ func (s *RockService) ListTokens(ctx context.Context, page, pageSize int, status
}
var tokens []model.RockToken
err := db.Order("id DESC").
// 按包 ID 倒序(未打包的手工采购记录排在最后),同包内按本地 ID 倒序,保证分页稳定
err := db.Order("package_id DESC NULLS LAST").
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))