feat:采购日志
This commit is contained in:
@@ -0,0 +1,53 @@
|
||||
-- 019_rock_purchase_logs.sql
|
||||
-- 采购流水:每一次进入采购流程的尝试(含降级、跳过、失败)都留一条主记录,
|
||||
-- 每个上游接口调用在阶段表里保留原始请求/响应,便于对账与排障。
|
||||
-- 注意:只记录请求体,不记录任何请求头,避免 Bearer Token 落库。
|
||||
|
||||
CREATE TABLE IF NOT EXISTS rock_purchase_logs (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
trigger_type VARCHAR(16) NOT NULL DEFAULT 'auto', -- auto / manual
|
||||
status VARCHAR(16) NOT NULL DEFAULT 'running',-- running / success / partial / failed / skipped
|
||||
requested_count INTEGER NOT NULL DEFAULT 0, -- 原计划(批次值或手动输入)
|
||||
planned_count INTEGER NOT NULL DEFAULT 0, -- 降级后的实际计划数
|
||||
claimed_count INTEGER NOT NULL DEFAULT 0, -- 上游实际返回数
|
||||
imported_count INTEGER NOT NULL DEFAULT 0, -- 本地实际入库数
|
||||
upstream_available INTEGER NOT NULL DEFAULT 0, -- 采购时上游可用库存
|
||||
points_before NUMERIC(12,2) NOT NULL DEFAULT 0.00, -- 采购前积分
|
||||
unit_cost NUMERIC(12,2) NOT NULL DEFAULT 0.00, -- 采购单价
|
||||
max_by_points INTEGER NOT NULL DEFAULT 0, -- 积分可买上限
|
||||
max_by_stock INTEGER NOT NULL DEFAULT 0, -- 上游库存可买上限
|
||||
limit_reason VARCHAR(32) NOT NULL DEFAULT '', -- points / upstream_stock / points_and_stock
|
||||
points_source VARCHAR(16) NOT NULL DEFAULT '', -- live / snapshot
|
||||
stage VARCHAR(32) NOT NULL DEFAULT '', -- me / dashboard / claim / package / detail / import
|
||||
error_message TEXT NOT NULL DEFAULT '',
|
||||
package_id BIGINT,
|
||||
share_code VARCHAR(64) NOT NULL DEFAULT '',
|
||||
duration_ms INTEGER NOT NULL DEFAULT 0,
|
||||
started_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
finished_at TIMESTAMPTZ,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_rock_purchase_logs_created_at
|
||||
ON rock_purchase_logs (created_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_rock_purchase_logs_trigger_status
|
||||
ON rock_purchase_logs (trigger_type, status);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS rock_purchase_stages (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
log_id BIGINT NOT NULL REFERENCES rock_purchase_logs(id) ON DELETE CASCADE,
|
||||
seq INTEGER NOT NULL DEFAULT 0,
|
||||
stage VARCHAR(32) NOT NULL,
|
||||
method VARCHAR(8) NOT NULL DEFAULT '',
|
||||
url TEXT NOT NULL DEFAULT '',
|
||||
request_body TEXT NOT NULL DEFAULT '',
|
||||
http_status INTEGER NOT NULL DEFAULT 0,
|
||||
response_body TEXT NOT NULL DEFAULT '',
|
||||
truncated BOOLEAN NOT NULL DEFAULT false,
|
||||
duration_ms INTEGER NOT NULL DEFAULT 0,
|
||||
error_message TEXT NOT NULL DEFAULT '',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_rock_purchase_stages_log_id
|
||||
ON rock_purchase_stages (log_id, seq);
|
||||
@@ -2,6 +2,7 @@ package handler
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
@@ -100,12 +101,70 @@ func (h *RockHandler) PurchaseTokens(c *gin.Context) {
|
||||
response.BadRequest(c, err.Error())
|
||||
return
|
||||
}
|
||||
summary := h.rockSvc.LastPurchaseSummary()
|
||||
message := "采购成功,已批量入库"
|
||||
planned := len(tokens)
|
||||
limitReason := ""
|
||||
if summary != nil {
|
||||
planned = summary.PlannedCount
|
||||
limitReason = summary.LimitReason
|
||||
switch summary.LimitReason {
|
||||
case "upstream_stock":
|
||||
message = fmt.Sprintf("上游货源不足,已按 %d 个采购", summary.PlannedCount)
|
||||
case "points":
|
||||
message = fmt.Sprintf("积分不足,已按 %d 个采购", summary.PlannedCount)
|
||||
case "points_and_stock":
|
||||
message = fmt.Sprintf("积分与上游库存不足,已按 %d 个采购", summary.PlannedCount)
|
||||
case "points_snapshot":
|
||||
message = fmt.Sprintf("积分不足(按快照估算),已按 %d 个采购", summary.PlannedCount)
|
||||
}
|
||||
}
|
||||
response.OK(c, gin.H{
|
||||
"purchased_count": len(tokens),
|
||||
"message": "采购成功,已批量入库",
|
||||
"requested_count": req.Count,
|
||||
"planned_count": planned,
|
||||
"limit_reason": limitReason,
|
||||
"log_id": summaryLogID(summary),
|
||||
"message": message,
|
||||
})
|
||||
}
|
||||
|
||||
func summaryLogID(summary *service.PurchaseSummary) uint {
|
||||
if summary == nil {
|
||||
return 0
|
||||
}
|
||||
return summary.LogID
|
||||
}
|
||||
|
||||
// ListPurchaseLogs 分页查询采购流水
|
||||
func (h *RockHandler) ListPurchaseLogs(c *gin.Context) {
|
||||
page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
|
||||
pageSize, _ := strconv.Atoi(c.DefaultQuery("page_size", "20"))
|
||||
triggerType := strings.TrimSpace(c.Query("trigger_type"))
|
||||
status := strings.TrimSpace(c.Query("status"))
|
||||
list, total, err := h.rockSvc.ListPurchaseLogs(c.Request.Context(), page, pageSize, triggerType, status)
|
||||
if err != nil {
|
||||
response.ServerError(c, err.Error())
|
||||
return
|
||||
}
|
||||
response.OK(c, gin.H{"list": list, "total": total})
|
||||
}
|
||||
|
||||
// GetPurchaseLog 查询单次采购详情(含各阶段原始报文)
|
||||
func (h *RockHandler) GetPurchaseLog(c *gin.Context) {
|
||||
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
|
||||
if err != nil || id == 0 {
|
||||
response.BadRequest(c, "无效的日志 ID")
|
||||
return
|
||||
}
|
||||
entry, stages, err := h.rockSvc.GetPurchaseLog(c.Request.Context(), uint(id))
|
||||
if err != nil {
|
||||
response.NotFound(c, "采购日志不存在")
|
||||
return
|
||||
}
|
||||
response.OK(c, gin.H{"log": entry, "stages": stages})
|
||||
}
|
||||
|
||||
// ListTokens 分页获取 Token 库存列表
|
||||
func (h *RockHandler) ListTokens(c *gin.Context) {
|
||||
page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
|
||||
|
||||
@@ -100,3 +100,56 @@ type RockToken struct {
|
||||
func (RockToken) TableName() string {
|
||||
return "rock_tokens"
|
||||
}
|
||||
|
||||
// RockPurchaseLog 是一次采购尝试的主记录。
|
||||
// 每一次真正进入采购流程的尝试都会落一条,包含降级原因与各阶段汇总。
|
||||
type RockPurchaseLog struct {
|
||||
ID uint `gorm:"primaryKey" json:"id"`
|
||||
TriggerType string `gorm:"size:16;not null;default:'auto';index:idx_rock_purchase_logs_trigger_status" json:"trigger_type"` // auto / manual
|
||||
Status string `gorm:"size:16;not null;default:'running'" json:"status"` // running / success / partial / failed / skipped
|
||||
RequestedCount int `gorm:"not null;default:0" json:"requested_count"`
|
||||
PlannedCount int `gorm:"not null;default:0" json:"planned_count"`
|
||||
ClaimedCount int `gorm:"not null;default:0" json:"claimed_count"`
|
||||
ImportedCount int `gorm:"not null;default:0" json:"imported_count"`
|
||||
UpstreamAvailable int `gorm:"not null;default:0" json:"upstream_available"`
|
||||
PointsBefore float64 `gorm:"type:numeric(12,2);not null;default:0" json:"points_before"`
|
||||
UnitCost float64 `gorm:"type:numeric(12,2);not null;default:0" json:"unit_cost"`
|
||||
MaxByPoints int `gorm:"not null;default:0" json:"max_by_points"`
|
||||
MaxByStock int `gorm:"not null;default:0" json:"max_by_stock"`
|
||||
LimitReason string `gorm:"size:32;not null;default:''" json:"limit_reason"` // points / upstream_stock / points_and_stock
|
||||
PointsSource string `gorm:"size:16;not null;default:''" json:"points_source"` // live / snapshot
|
||||
Stage string `gorm:"size:32;not null;default:''" json:"stage"`
|
||||
ErrorMessage string `gorm:"type:text;not null;default:''" json:"error_message"`
|
||||
PackageID *int64 `json:"package_id,omitempty"`
|
||||
ShareCode string `gorm:"size:64;not null;default:''" json:"share_code"`
|
||||
DurationMs int `gorm:"not null;default:0" json:"duration_ms"`
|
||||
StartedAt time.Time `json:"started_at"`
|
||||
FinishedAt *time.Time `json:"finished_at,omitempty"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
func (RockPurchaseLog) TableName() string {
|
||||
return "rock_purchase_logs"
|
||||
}
|
||||
|
||||
// RockPurchaseStage 记录一次采购中单个上游调用的原始请求与响应。
|
||||
// ResponseBody 保存上游返回的原始 JSON 原文;超过 512KB 时截断并置 Truncated。
|
||||
type RockPurchaseStage struct {
|
||||
ID uint `gorm:"primaryKey" json:"id"`
|
||||
LogID uint `gorm:"not null;index:idx_rock_purchase_stages_log_id" json:"log_id"`
|
||||
Seq int `gorm:"not null;default:0" json:"seq"`
|
||||
Stage string `gorm:"size:32;not null" json:"stage"`
|
||||
Method string `gorm:"size:8;not null;default:''" json:"method"`
|
||||
URL string `gorm:"type:text;not null;default:''" json:"url"`
|
||||
RequestBody string `gorm:"type:text;not null;default:''" json:"request_body"`
|
||||
HTTPStatus int `gorm:"not null;default:0" json:"http_status"`
|
||||
ResponseBody string `gorm:"type:text;not null;default:''" json:"response_body"`
|
||||
Truncated bool `gorm:"not null;default:false" json:"truncated"`
|
||||
DurationMs int `gorm:"not null;default:0" json:"duration_ms"`
|
||||
ErrorMessage string `gorm:"type:text;not null;default:''" json:"error_message"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
func (RockPurchaseStage) TableName() string {
|
||||
return "rock_purchase_stages"
|
||||
}
|
||||
|
||||
@@ -289,6 +289,8 @@ func Setup(h *Handlers) *gin.Engine {
|
||||
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/purchase-logs", h.Rock.ListPurchaseLogs)
|
||||
platform.GET("/platform/rock/purchase-logs/:id", h.Rock.GetPurchaseLog)
|
||||
platform.GET("/platform/rock/tokens", h.Rock.ListTokens)
|
||||
platform.POST("/platform/rock/tokens/:id/expected-uid", h.Rock.CorrectTokenUID)
|
||||
platform.POST("/platform/rock/tokens/:id/reset", h.Rock.ResetToken)
|
||||
|
||||
@@ -374,6 +374,7 @@ func (r *RockPackageCreateResponse) ResolvePackage() (int64, string) {
|
||||
// CreatePackage 把一批已提取 Token 打包成公开分享包(需 Bearer Token)。
|
||||
// 注意:上游未返回 share_code 时不会在这里报错,由调用方决定兜底策略。
|
||||
func (c *RockClient) CreatePackage(ctx context.Context, bearerToken string, ids []string, note string) (*RockPackageCreateResponse, error) {
|
||||
startedAt := time.Now()
|
||||
if len(ids) == 0 {
|
||||
return nil, fmt.Errorf("打包 ids 不能为空")
|
||||
}
|
||||
@@ -404,6 +405,7 @@ func (c *RockClient) CreatePackage(ctx context.Context, bearerToken string, ids
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取 packages 创建响应失败: %w", err)
|
||||
}
|
||||
traceRockRequest(ctx, "package", http.MethodPost, reqURL, jsonBytes, resp.StatusCode, body, time.Since(startedAt), err)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("上游 packages 创建接口返回错误码 %d: %s", resp.StatusCode, string(body))
|
||||
}
|
||||
@@ -420,6 +422,7 @@ func (c *RockClient) CreatePackage(ctx context.Context, bearerToken string, ids
|
||||
|
||||
// GetPackageDetail 按 share_code 查询打包详情(公开接口,无需 Bearer)。
|
||||
func (c *RockClient) GetPackageDetail(ctx context.Context, shareCode string) (*RockPackageDetailResponse, error) {
|
||||
startedAt := time.Now()
|
||||
shareCode = strings.TrimSpace(shareCode)
|
||||
if shareCode == "" {
|
||||
return nil, fmt.Errorf("share_code 不能为空")
|
||||
@@ -441,6 +444,7 @@ func (c *RockClient) GetPackageDetail(ctx context.Context, shareCode string) (*R
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取 package 详情响应失败: %w", err)
|
||||
}
|
||||
traceRockRequest(ctx, "detail", http.MethodGet, reqURL, nil, resp.StatusCode, body, time.Since(startedAt), err)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("上游 package 详情接口返回错误码 %d: %s", resp.StatusCode, string(body))
|
||||
}
|
||||
@@ -456,6 +460,7 @@ func (c *RockClient) GetPackageDetail(ctx context.Context, shareCode string) (*R
|
||||
|
||||
// GetMe 获取商户个人信息及积分
|
||||
func (c *RockClient) GetMe(ctx context.Context, bearerToken string) (*RockMeResponse, error) {
|
||||
startedAt := time.Now()
|
||||
reqURL := fmt.Sprintf("%s/api/me", c.BaseURL())
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
|
||||
if err != nil {
|
||||
@@ -474,6 +479,7 @@ func (c *RockClient) GetMe(ctx context.Context, bearerToken string) (*RockMeResp
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取响应失败: %w", err)
|
||||
}
|
||||
traceRockRequest(ctx, "me", http.MethodGet, reqURL, nil, resp.StatusCode, body, time.Since(startedAt), err)
|
||||
|
||||
if resp.StatusCode == http.StatusUnauthorized {
|
||||
return nil, fmt.Errorf("上游 Bearer Token 已失效或未授权 (401)")
|
||||
@@ -532,6 +538,7 @@ func (r *RockDashboardResponse) ResolveBlueTokenCost() float64 {
|
||||
|
||||
// GetDashboard 获取上游可用库存与实时采购单价(需 Bearer Token)。
|
||||
func (c *RockClient) GetDashboard(ctx context.Context, bearerToken string) (*RockDashboardResponse, error) {
|
||||
startedAt := time.Now()
|
||||
reqURL := fmt.Sprintf("%s/api/user/dashboard", c.BaseURL())
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
|
||||
if err != nil {
|
||||
@@ -552,6 +559,7 @@ func (c *RockClient) GetDashboard(ctx context.Context, bearerToken string) (*Roc
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取 dashboard 响应失败: %w", err)
|
||||
}
|
||||
traceRockRequest(ctx, "dashboard", http.MethodGet, reqURL, nil, resp.StatusCode, body, time.Since(startedAt), err)
|
||||
if resp.StatusCode == http.StatusUnauthorized {
|
||||
return nil, fmt.Errorf("上游 Bearer Token 已失效或未授权 (401)")
|
||||
}
|
||||
@@ -570,6 +578,7 @@ func (c *RockClient) GetDashboard(ctx context.Context, bearerToken string) (*Roc
|
||||
|
||||
// ClaimTokens 批量提取采购 Token
|
||||
func (c *RockClient) ClaimTokens(ctx context.Context, bearerToken string, color string, count int) ([]RockClaimedToken, error) {
|
||||
startedAt := time.Now()
|
||||
if count <= 0 {
|
||||
return nil, fmt.Errorf("提取数量必须大于 0")
|
||||
}
|
||||
@@ -600,6 +609,7 @@ func (c *RockClient) ClaimTokens(ctx context.Context, bearerToken string, color
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取 claim 响应失败: %w", err)
|
||||
}
|
||||
traceRockRequest(ctx, "claim", http.MethodPost, reqURL, jsonBytes, resp.StatusCode, body, time.Since(startedAt), err)
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("上游 claim 接口返回错误码 %d: %s", resp.StatusCode, string(body))
|
||||
|
||||
@@ -101,7 +101,7 @@ func TestRefreshDashboard_UpdatesChangedCost(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestPurchaseTokens_BlocksWhenUpstreamStockInsufficient(t *testing.T) {
|
||||
func TestPurchaseTokens_DowngradesWhenUpstreamStockInsufficient(t *testing.T) {
|
||||
db := newRockTestDB(t)
|
||||
var claimCalls int32
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -124,23 +124,30 @@ func TestPurchaseTokens_BlocksWhenUpstreamStockInsufficient(t *testing.T) {
|
||||
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)
|
||||
tokens, err := svc.PurchaseTokens(context.Background(), 2, false)
|
||||
if err != nil {
|
||||
t.Fatalf("上游库存不足时应降级采购而非失败,实际 err=%v", err)
|
||||
}
|
||||
if calls := atomic.LoadInt32(&claimCalls); calls != 0 {
|
||||
t.Fatalf("库存不足时不应调用上游 claim,实际调用 %d 次", calls)
|
||||
if calls := atomic.LoadInt32(&claimCalls); calls != 1 {
|
||||
t.Fatalf("降级后应调用上游 claim 1 次,实际 %d 次", calls)
|
||||
}
|
||||
if len(tokens) != 1 {
|
||||
t.Fatalf("期望按库存上限入库 1 个 Token,实际 %d 个", len(tokens))
|
||||
}
|
||||
summary := svc.LastPurchaseSummary()
|
||||
if summary == nil {
|
||||
t.Fatal("缺少采购摘要")
|
||||
}
|
||||
if summary.PlannedCount != 1 || summary.LimitReason != "upstream_stock" {
|
||||
t.Fatalf("降级信息不符: planned=%d reason=%s", summary.PlannedCount, summary.LimitReason)
|
||||
}
|
||||
|
||||
var count int64
|
||||
db.Model(&model.RockToken{}).Count(&count)
|
||||
if count != 0 {
|
||||
t.Fatalf("库存不足时不应入库任何 Token,实际 %d 条", count)
|
||||
var entry model.RockPurchaseLog
|
||||
if err := db.Order("id DESC").First(&entry).Error; err != nil {
|
||||
t.Fatalf("查询采购日志失败: %v", err)
|
||||
}
|
||||
|
||||
cfg, _ := svc.GetConfig(context.Background())
|
||||
if !strings.Contains(cfg.LastError, "上游可用库存不足") {
|
||||
t.Fatalf("last_error 应记录库存不足,实际 %q", cfg.LastError)
|
||||
if entry.Status != "partial" || entry.MaxByStock != 1 {
|
||||
t.Fatalf("采购日志不符: status=%s max_by_stock=%d", entry.Status, entry.MaxByStock)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,242 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"affiliate_dash/internal/model"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// rockPurchaseStageMaxBytes 单个阶段原始响应/请求的最大保存体积。
|
||||
// 超过后截断并标记 truncated,避免异常大响应撑爆数据库。
|
||||
const rockPurchaseStageMaxBytes = 512 * 1024
|
||||
|
||||
// PurchaseSummary 是最近一次采购的可展示摘要,供手动采购接口回显降级结果。
|
||||
type PurchaseSummary struct {
|
||||
LogID uint `json:"log_id"`
|
||||
TriggerType string `json:"trigger_type"`
|
||||
Status string `json:"status"`
|
||||
RequestedCount int `json:"requested_count"`
|
||||
PlannedCount int `json:"planned_count"`
|
||||
ClaimedCount int `json:"claimed_count"`
|
||||
ImportedCount int `json:"imported_count"`
|
||||
LimitReason string `json:"limit_reason"`
|
||||
ErrorMessage string `json:"error_message"`
|
||||
}
|
||||
|
||||
// purchaseRecorder 负责一次采购的日志生命周期:
|
||||
// 开始时插入主记录(running),每个上游调用写一条阶段记录,结束时更新主记录。
|
||||
// 所有写入失败都只记日志,绝不阻断采购主流程。
|
||||
type purchaseRecorder struct {
|
||||
db *gorm.DB
|
||||
logID uint
|
||||
seq int
|
||||
requested int
|
||||
planned int
|
||||
claimed int
|
||||
imported int
|
||||
limitReason string
|
||||
pointsSource string
|
||||
stage string
|
||||
errMessage string
|
||||
degraded bool
|
||||
|
||||
upstreamAvailable int
|
||||
pointsBefore float64
|
||||
unitCost float64
|
||||
maxByPoints int
|
||||
maxByStock int
|
||||
packageID *int64
|
||||
shareCode string
|
||||
}
|
||||
|
||||
func newPurchaseRecorder(db *gorm.DB, triggerType string, requested int, startedAt time.Time) *purchaseRecorder {
|
||||
rec := &purchaseRecorder{db: db, requested: requested}
|
||||
entry := model.RockPurchaseLog{
|
||||
TriggerType: triggerType,
|
||||
Status: "running",
|
||||
RequestedCount: requested,
|
||||
StartedAt: startedAt,
|
||||
CreatedAt: startedAt,
|
||||
}
|
||||
if err := db.WithContext(context.Background()).Create(&entry).Error; err != nil {
|
||||
log.Printf("[RockService] 创建采购日志失败,本次采购不记录流水: %v", err)
|
||||
return rec
|
||||
}
|
||||
rec.logID = entry.ID
|
||||
return rec
|
||||
}
|
||||
|
||||
// addStage 记录一次上游调用的原始请求与响应。
|
||||
func (r *purchaseRecorder) addStage(stage, method, url string, reqBody []byte, status int, respBody []byte, dur time.Duration, callErr error) {
|
||||
if r == nil || r.logID == 0 {
|
||||
return
|
||||
}
|
||||
r.seq++
|
||||
responseText, truncated := truncateStageBody(respBody)
|
||||
requestText, _ := truncateStageBody(reqBody)
|
||||
|
||||
errText := ""
|
||||
if callErr != nil {
|
||||
errText = callErr.Error()
|
||||
}
|
||||
if errText == "" && status >= 400 {
|
||||
errText = fmt.Sprintf("上游返回 HTTP %d", status)
|
||||
}
|
||||
if errText != "" && r.stage == "" {
|
||||
r.stage = stage
|
||||
r.errMessage = truncatePurchaseError(errText)
|
||||
}
|
||||
|
||||
entry := model.RockPurchaseStage{
|
||||
LogID: r.logID,
|
||||
Seq: r.seq,
|
||||
Stage: stage,
|
||||
Method: method,
|
||||
URL: url,
|
||||
RequestBody: requestText,
|
||||
HTTPStatus: status,
|
||||
ResponseBody: responseText,
|
||||
Truncated: truncated,
|
||||
DurationMs: int(dur.Milliseconds()),
|
||||
ErrorMessage: truncatePurchaseError(errText),
|
||||
CreatedAt: time.Now(),
|
||||
}
|
||||
if err := r.db.WithContext(context.Background()).Create(&entry).Error; err != nil {
|
||||
log.Printf("[RockService] 写入采购阶段日志失败 stage=%s: %v", stage, err)
|
||||
}
|
||||
}
|
||||
|
||||
// markDegraded 标记本次采购虽有产出但存在降级/兜底行为,最终状态记为 partial。
|
||||
func (r *purchaseRecorder) markDegraded(reason string) {
|
||||
if r == nil {
|
||||
return
|
||||
}
|
||||
r.degraded = true
|
||||
if r.errMessage == "" {
|
||||
r.errMessage = truncatePurchaseError(reason)
|
||||
}
|
||||
}
|
||||
|
||||
// finish 更新主记录并返回摘要。
|
||||
func (r *purchaseRecorder) finish(tokens []model.RockToken, runErr error, duration time.Duration) *PurchaseSummary {
|
||||
if r == nil {
|
||||
return nil
|
||||
}
|
||||
if r.imported == 0 && len(tokens) > 0 {
|
||||
r.imported = len(tokens)
|
||||
}
|
||||
status := "success"
|
||||
switch {
|
||||
case r.planned <= 0:
|
||||
status = "skipped"
|
||||
case runErr != nil:
|
||||
status = "failed"
|
||||
case r.degraded || r.planned < r.requested || (r.claimed > 0 && r.imported < r.claimed):
|
||||
status = "partial"
|
||||
}
|
||||
if runErr != nil && r.errMessage == "" {
|
||||
r.errMessage = truncatePurchaseError(runErr.Error())
|
||||
}
|
||||
|
||||
finished := time.Now()
|
||||
summary := &PurchaseSummary{
|
||||
LogID: r.logID,
|
||||
Status: status,
|
||||
RequestedCount: r.requested,
|
||||
PlannedCount: r.planned,
|
||||
ClaimedCount: r.claimed,
|
||||
ImportedCount: r.imported,
|
||||
LimitReason: r.limitReason,
|
||||
ErrorMessage: r.errMessage,
|
||||
}
|
||||
if r.logID == 0 {
|
||||
return summary
|
||||
}
|
||||
updates := map[string]interface{}{
|
||||
"status": status,
|
||||
"planned_count": r.planned,
|
||||
"claimed_count": r.claimed,
|
||||
"imported_count": r.imported,
|
||||
"upstream_available": r.upstreamAvailable,
|
||||
"points_before": r.pointsBefore,
|
||||
"unit_cost": r.unitCost,
|
||||
"max_by_points": r.maxByPoints,
|
||||
"max_by_stock": r.maxByStock,
|
||||
"limit_reason": r.limitReason,
|
||||
"points_source": r.pointsSource,
|
||||
"stage": r.stage,
|
||||
"error_message": r.errMessage,
|
||||
"package_id": r.packageID,
|
||||
"share_code": r.shareCode,
|
||||
"duration_ms": int(duration.Milliseconds()),
|
||||
"finished_at": finished,
|
||||
}
|
||||
if err := r.db.WithContext(context.Background()).Model(&model.RockPurchaseLog{}).
|
||||
Where("id = ?", r.logID).Updates(updates).Error; err != nil {
|
||||
log.Printf("[RockService] 更新采购日志失败 id=%d: %v", r.logID, err)
|
||||
}
|
||||
return summary
|
||||
}
|
||||
|
||||
// truncateStageBody 把原始请求/响应限制在 512KB 内,保证截断后仍是合法 UTF-8。
|
||||
func truncateStageBody(raw []byte) (string, bool) {
|
||||
if len(raw) == 0 {
|
||||
return "", false
|
||||
}
|
||||
if len(raw) <= rockPurchaseStageMaxBytes {
|
||||
return string(raw), false
|
||||
}
|
||||
cut := raw[:rockPurchaseStageMaxBytes]
|
||||
return strings.ToValidUTF8(string(cut), "?"), true
|
||||
}
|
||||
|
||||
// LastPurchaseSummary 返回最近一次采购的摘要(包含降级原因),供手动采购接口回显。
|
||||
func (s *RockService) LastPurchaseSummary() *PurchaseSummary {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.lastSummary
|
||||
}
|
||||
|
||||
// ListPurchaseLogs 分页查询采购流水(不返回阶段大字段)。
|
||||
func (s *RockService) ListPurchaseLogs(ctx context.Context, page, pageSize int, triggerType, status string) ([]model.RockPurchaseLog, int64, error) {
|
||||
if page <= 0 {
|
||||
page = 1
|
||||
}
|
||||
if pageSize <= 0 || pageSize > 200 {
|
||||
pageSize = 20
|
||||
}
|
||||
db := s.db.WithContext(ctx).Model(&model.RockPurchaseLog{})
|
||||
if t := strings.TrimSpace(triggerType); t != "" {
|
||||
db = db.Where("trigger_type = ?", t)
|
||||
}
|
||||
if st := strings.TrimSpace(status); st != "" {
|
||||
db = db.Where("status = ?", st)
|
||||
}
|
||||
var total int64
|
||||
if err := db.Count(&total).Error; err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
var list []model.RockPurchaseLog
|
||||
if err := db.Order("id DESC").Offset((page - 1) * pageSize).Limit(pageSize).Find(&list).Error; err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return list, total, nil
|
||||
}
|
||||
|
||||
// GetPurchaseLog 查询单次采购详情,包含全部阶段原始报文。
|
||||
func (s *RockService) GetPurchaseLog(ctx context.Context, id uint) (*model.RockPurchaseLog, []model.RockPurchaseStage, error) {
|
||||
var entry model.RockPurchaseLog
|
||||
if err := s.db.WithContext(ctx).First(&entry, id).Error; err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
var stages []model.RockPurchaseStage
|
||||
if err := s.db.WithContext(ctx).Where("log_id = ?", id).Order("seq ASC, id ASC").Find(&stages).Error; err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return &entry, stages, nil
|
||||
}
|
||||
@@ -0,0 +1,239 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"math"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"affiliate_dash/internal/model"
|
||||
)
|
||||
|
||||
// newPurchaseMockServer 构造覆盖采购全链路的上游 mock,返回服务与 claim 调用计数。
|
||||
func newPurchaseMockServer(t *testing.T, points string, stock int, cost string, claimStatus int) (*httptest.Server, *int32) {
|
||||
t.Helper()
|
||||
var claimCalls int32
|
||||
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/me":
|
||||
_ = json.NewEncoder(w).Encode(map[string]interface{}{
|
||||
"user": map[string]interface{}{"username": "tester", "points": points},
|
||||
})
|
||||
case "/api/user/dashboard":
|
||||
writeRockDashboardJSON(w, stock, cost)
|
||||
case "/api/user/tokens/claim":
|
||||
atomic.AddInt32(&claimCalls, 1)
|
||||
if claimStatus != http.StatusOK {
|
||||
w.WriteHeader(claimStatus)
|
||||
_, _ = w.Write([]byte(`{"error":"claim down"}`))
|
||||
return
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(map[string]interface{}{
|
||||
"tokens": []map[string]interface{}{
|
||||
{"id": 60001, "token": "tk-1", "color": "black", "region": "Q/V可用", "status": "using", "redeem_code": "rc-1"},
|
||||
},
|
||||
})
|
||||
case "/api/user/packages":
|
||||
_ = json.NewEncoder(w).Encode(map[string]interface{}{
|
||||
"package": map[string]interface{}{"id": 8001, "share_code": "share-log", "token_count": 1},
|
||||
})
|
||||
case "/api/package/share-log":
|
||||
_ = json.NewEncoder(w).Encode(map[string]interface{}{
|
||||
"package": map[string]interface{}{"id": 8001, "share_code": "share-log", "token_count": 1},
|
||||
"tokens": []map[string]interface{}{
|
||||
{"id": 60001, "token": "tk-1", "color": "black", "region": "Q/V可用", "status": "using", "redeem_code": "rc-1"},
|
||||
},
|
||||
})
|
||||
default:
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
}
|
||||
}))
|
||||
return server, &claimCalls
|
||||
}
|
||||
|
||||
func TestPurchaseTokens_DowngradesByPoints(t *testing.T) {
|
||||
db := newRockTestDB(t)
|
||||
server, claimCalls := newPurchaseMockServer(t, "30.0", 100, "20.0", http.StatusOK)
|
||||
defer server.Close()
|
||||
|
||||
svc := NewRockService(db, NewRockClient(server.URL))
|
||||
setRockTestBearer(t, svc, "bearer-test")
|
||||
|
||||
tokens, err := svc.PurchaseTokens(context.Background(), 2, false)
|
||||
if err != nil {
|
||||
t.Fatalf("积分不足时应降级采购而非失败,实际 err=%v", err)
|
||||
}
|
||||
if len(tokens) != 1 {
|
||||
t.Fatalf("30 点 / 20 点单价期望采购 1 个,实际 %d 个", len(tokens))
|
||||
}
|
||||
if calls := atomic.LoadInt32(claimCalls); calls != 1 {
|
||||
t.Fatalf("应调用上游 claim 1 次,实际 %d 次", calls)
|
||||
}
|
||||
|
||||
summary := svc.LastPurchaseSummary()
|
||||
if summary == nil || summary.PlannedCount != 1 || summary.LimitReason != "points" {
|
||||
t.Fatalf("降级信息不符: %+v", summary)
|
||||
}
|
||||
|
||||
var entry model.RockPurchaseLog
|
||||
if err := db.Order("id DESC").First(&entry).Error; err != nil {
|
||||
t.Fatalf("查询采购日志失败: %v", err)
|
||||
}
|
||||
if entry.Status != "partial" {
|
||||
t.Fatalf("降级成交应记为 partial,实际 %s", entry.Status)
|
||||
}
|
||||
if entry.MaxByPoints != 1 || entry.MaxByStock != 100 {
|
||||
t.Fatalf("上限快照不符: points=%d stock=%d", entry.MaxByPoints, entry.MaxByStock)
|
||||
}
|
||||
if math.Abs(entry.PointsBefore-30) > 1e-9 || math.Abs(entry.UnitCost-20) > 1e-9 {
|
||||
t.Fatalf("积分/单价快照不符: points=%v cost=%v", entry.PointsBefore, entry.UnitCost)
|
||||
}
|
||||
if entry.PointsSource != "live" {
|
||||
t.Fatalf("应使用实时积分,实际来源 %s", entry.PointsSource)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPurchaseTokens_SkipsWhenBudgetZero(t *testing.T) {
|
||||
db := newRockTestDB(t)
|
||||
server, claimCalls := newPurchaseMockServer(t, "10.0", 100, "20.0", http.StatusOK)
|
||||
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 entry model.RockPurchaseLog
|
||||
if err := db.Order("id DESC").First(&entry).Error; err != nil {
|
||||
t.Fatalf("查询采购日志失败: %v", err)
|
||||
}
|
||||
if entry.Status != "skipped" || entry.PlannedCount != 0 || entry.LimitReason != "points" {
|
||||
t.Fatalf("跳过记录不符: status=%s planned=%d reason=%s", entry.Status, entry.PlannedCount, entry.LimitReason)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPurchaseTokens_DowngradesByPointsAndStock(t *testing.T) {
|
||||
db := newRockTestDB(t)
|
||||
server, _ := newPurchaseMockServer(t, "30.0", 1, "20.0", http.StatusOK)
|
||||
defer server.Close()
|
||||
|
||||
svc := NewRockService(db, NewRockClient(server.URL))
|
||||
setRockTestBearer(t, svc, "bearer-test")
|
||||
|
||||
tokens, err := svc.PurchaseTokens(context.Background(), 5, false)
|
||||
if err != nil {
|
||||
t.Fatalf("双重限制时应降级采购,实际 err=%v", err)
|
||||
}
|
||||
if len(tokens) != 1 {
|
||||
t.Fatalf("积分与库存都只够 1 个,实际入库 %d 个", len(tokens))
|
||||
}
|
||||
summary := svc.LastPurchaseSummary()
|
||||
if summary == nil || summary.LimitReason != "points_and_stock" || summary.PlannedCount != 1 {
|
||||
t.Fatalf("双重限制信息不符: %+v", summary)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPurchaseTokens_PersistsRawResponse(t *testing.T) {
|
||||
db := newRockTestDB(t)
|
||||
server, _ := newPurchaseMockServer(t, "1000.0", 100, "20.0", http.StatusOK)
|
||||
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))
|
||||
}
|
||||
summary := svc.LastPurchaseSummary()
|
||||
if summary == nil || summary.LogID == 0 {
|
||||
t.Fatal("缺少采购日志 ID")
|
||||
}
|
||||
|
||||
_, stages, err := svc.GetPurchaseLog(context.Background(), summary.LogID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询采购阶段失败: %v", err)
|
||||
}
|
||||
if len(stages) < 5 {
|
||||
t.Fatalf("应记录 me/dashboard/claim/package/detail 至少 5 个阶段,实际 %d 个", len(stages))
|
||||
}
|
||||
var claimStage *model.RockPurchaseStage
|
||||
for i := range stages {
|
||||
if stages[i].Stage == "claim" {
|
||||
claimStage = &stages[i]
|
||||
break
|
||||
}
|
||||
}
|
||||
if claimStage == nil {
|
||||
t.Fatal("缺少 claim 阶段记录")
|
||||
}
|
||||
if claimStage.HTTPStatus != http.StatusOK {
|
||||
t.Fatalf("claim 阶段状态码应为 200,实际 %d", claimStage.HTTPStatus)
|
||||
}
|
||||
if !strings.Contains(claimStage.ResponseBody, "60001") {
|
||||
t.Fatalf("原始响应未保留,实际 %q", claimStage.ResponseBody)
|
||||
}
|
||||
if !strings.Contains(claimStage.RequestBody, "\"count\"") {
|
||||
t.Fatalf("请求体未保留,实际 %q", claimStage.RequestBody)
|
||||
}
|
||||
if strings.Contains(claimStage.RequestBody, "bearer-test") {
|
||||
t.Fatal("请求体不应包含 Bearer Token")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPurchaseTokens_RecordsFailedStage(t *testing.T) {
|
||||
db := newRockTestDB(t)
|
||||
server, _ := newPurchaseMockServer(t, "1000.0", 100, "20.0", http.StatusInternalServerError)
|
||||
defer server.Close()
|
||||
|
||||
svc := NewRockService(db, NewRockClient(server.URL))
|
||||
setRockTestBearer(t, svc, "bearer-test")
|
||||
|
||||
_, err := svc.PurchaseTokens(context.Background(), 1, false)
|
||||
if err == nil {
|
||||
t.Fatal("上游 claim 返回 500 时应当失败")
|
||||
}
|
||||
|
||||
var entry model.RockPurchaseLog
|
||||
if err := db.Order("id DESC").First(&entry).Error; err != nil {
|
||||
t.Fatalf("查询采购日志失败: %v", err)
|
||||
}
|
||||
if entry.Status != "failed" || entry.Stage != "claim" {
|
||||
t.Fatalf("失败记录不符: status=%s stage=%s", entry.Status, entry.Stage)
|
||||
}
|
||||
|
||||
_, stages, err := svc.GetPurchaseLog(context.Background(), entry.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询阶段失败: %v", err)
|
||||
}
|
||||
var failedStage *model.RockPurchaseStage
|
||||
for i := range stages {
|
||||
if stages[i].Stage == "claim" {
|
||||
failedStage = &stages[i]
|
||||
break
|
||||
}
|
||||
}
|
||||
if failedStage == nil {
|
||||
t.Fatal("缺少 claim 阶段记录")
|
||||
}
|
||||
if !strings.Contains(failedStage.ResponseBody, "claim down") {
|
||||
t.Fatalf("失败响应原文未保留,实际 %q", failedStage.ResponseBody)
|
||||
}
|
||||
if failedStage.ErrorMessage == "" {
|
||||
t.Fatal("失败阶段应记录错误信息")
|
||||
}
|
||||
}
|
||||
@@ -18,6 +18,10 @@ func setRockTestBearer(t *testing.T, svc *RockService, token string) {
|
||||
if _, err := svc.UpdateConfig(context.Background(), UpdateRockConfigReq{BearerToken: &bearer}); err != nil {
|
||||
t.Fatalf("设置 Bearer Token 失败: %v", err)
|
||||
}
|
||||
// 测试默认给足积分,避免降级逻辑把采购量算成 0;积分限制由专门用例覆盖。
|
||||
if err := svc.db.Model(&model.RockConfig{}).Where("id = 1").Update("points", 100000).Error; err != nil {
|
||||
t.Fatalf("设置测试积分失败: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPurchaseTokens_CreatesPackageAndImportsAvailable(t *testing.T) {
|
||||
|
||||
@@ -20,6 +20,7 @@ type RockService struct {
|
||||
mu sync.Mutex
|
||||
refreshCache sync.Map // key: upstreamTokenID string, value: rockRefreshCacheItem
|
||||
bindLinkCache sync.Map // key: token ID uint, value: rockBindLinkCacheItem
|
||||
lastSummary *PurchaseSummary
|
||||
}
|
||||
|
||||
func NewRockService(db *gorm.DB, client *RockClient) *RockService {
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"math"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -19,13 +20,26 @@ const (
|
||||
rockPurchaseNoteManualPrefix = "手动采购"
|
||||
)
|
||||
|
||||
// PurchaseTokens 执行采购并入库。
|
||||
// 流程:claim 批量提取 → 创建上游打包记录 → 拉取包详情 → 幂等导入本地(新记录 available)。
|
||||
// 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()
|
||||
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")
|
||||
}
|
||||
@@ -54,25 +68,80 @@ func (s *RockService) PurchaseTokens(ctx context.Context, count int, isAuto bool
|
||||
}
|
||||
s.updatePurchaseAttempt(ctx, cfg.ID, now)
|
||||
|
||||
// 采购前先刷新上游 dashboard:校验可用库存,同时同步最新采购单价。
|
||||
// 把采购追踪器挂到 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
|
||||
}
|
||||
if dashboard.AvailableBlack < count {
|
||||
err := fmt.Errorf("上游可用库存不足:当前 %d 个,本次需要 %d 个", dashboard.AvailableBlack, count)
|
||||
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
|
||||
}
|
||||
|
||||
claimed, err := s.client.ClaimTokens(ctx, cfg.BearerToken, model.RockColorBlack, count)
|
||||
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())
|
||||
@@ -113,31 +182,39 @@ func (s *RockService) PurchaseTokens(ctx context.Context, count int, isAuto bool
|
||||
pkgErr = errors.New("上游打包接口未返回内容")
|
||||
}
|
||||
if pkgErr != nil {
|
||||
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, 0, "", "purchase", now,
|
||||
"创建上游打包记录失败: "+pkgErr.Error())
|
||||
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 {
|
||||
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, shareCode, "package", now,
|
||||
fmt.Sprintf("包 %s 详情拉取失败: %v", shareCode, detailErr))
|
||||
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 {
|
||||
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, shareCode, "package", now,
|
||||
fmt.Sprintf("包 %s 导入失败: %v", shareCode, importErr))
|
||||
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 {
|
||||
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, shareCode, "package", now,
|
||||
fmt.Sprintf("包 %s 未返回有效 Token", shareCode))
|
||||
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())
|
||||
}
|
||||
|
||||
@@ -148,6 +225,7 @@ func (s *RockService) PurchaseTokens(ctx context.Context, count int, isAuto bool
|
||||
}
|
||||
dbTokens = s.buildClaimedTokens(claimed, packageID, shareCode, "package", now)
|
||||
}
|
||||
rec.imported = len(dbTokens)
|
||||
|
||||
if ensureErr == nil {
|
||||
s.recordPurchaseSuccess(ctx, cfg.ID)
|
||||
@@ -351,6 +429,23 @@ func truncatePurchaseError(msg string) string {
|
||||
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)
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// rockTraceKey 用于把采购追踪器挂到 context 上。
|
||||
// 采用 context 传递而不是客户端全局字段,避免并发采购时互相串数据。
|
||||
type rockTraceKey struct{}
|
||||
|
||||
// RockTrace 收集一次采购流程中所有上游 HTTP 调用的原始请求与响应。
|
||||
// OnStage 在客户端每次请求结束后同步调用,由采购日志记录器落库。
|
||||
type RockTrace struct {
|
||||
mu sync.Mutex
|
||||
OnStage func(stage, method, url string, reqBody []byte, status int, respBody []byte, dur time.Duration, err error)
|
||||
}
|
||||
|
||||
// WithRockTrace 返回带追踪器的 context;trace 为 nil 时原样返回。
|
||||
func WithRockTrace(ctx context.Context, trace *RockTrace) context.Context {
|
||||
if trace == nil {
|
||||
return ctx
|
||||
}
|
||||
return context.WithValue(ctx, rockTraceKey{}, trace)
|
||||
}
|
||||
|
||||
func traceFromContext(ctx context.Context) *RockTrace {
|
||||
if ctx == nil {
|
||||
return nil
|
||||
}
|
||||
trace, _ := ctx.Value(rockTraceKey{}).(*RockTrace)
|
||||
return trace
|
||||
}
|
||||
|
||||
// traceRockRequest 记录一次上游调用;没有追踪器时是空操作。
|
||||
func traceRockRequest(ctx context.Context, stage, method, url string, reqBody []byte, status int, respBody []byte, dur time.Duration, err error) {
|
||||
trace := traceFromContext(ctx)
|
||||
if trace == nil || trace.OnStage == nil {
|
||||
return
|
||||
}
|
||||
trace.mu.Lock()
|
||||
handler := trace.OnStage
|
||||
trace.mu.Unlock()
|
||||
handler(stage, method, url, reqBody, status, respBody, dur, err)
|
||||
}
|
||||
Reference in New Issue
Block a user