feat: 完善 Rock 发货异常闭环与绑定判定

This commit is contained in:
yml
2026-11-14 23:58:57 +08:00
parent 9353b439a7
commit 6a87fe5228
45 changed files with 3497 additions and 389 deletions
+1 -1
View File
@@ -161,7 +161,7 @@ func main() {
CORSAllowedOrigins: cfg.CORSAllowedOrigins,
LewanWebhook: handler.NewLewanWebhookHandler(fulfillmentSvc, db, cfg.LewanWebhookSecret, cfg.LewanApplicationID),
PlatformLewan: handler.NewPlatformLewanHandler(merchantSvc, lewanClient),
Rock: handler.NewRockHandler(rockSvc),
Rock: handler.NewRockHandler(rockSvc, deliverySvc),
}
go callbackSvc.Run(context.Background())
@@ -0,0 +1,34 @@
-- 029: Rock 发货异常事件表。
-- 原始请求/响应继续保留在 rock_delivery_logs;本表用于分类、筛选和人工处理。
CREATE TABLE IF NOT EXISTS rock_delivery_incidents (
id BIGSERIAL PRIMARY KEY,
token_id BIGINT NOT NULL,
order_no VARCHAR(128) NOT NULL DEFAULT '',
upstream_token_id VARCHAR(64) NOT NULL DEFAULT '',
category VARCHAR(40) NOT NULL,
status VARCHAR(16) NOT NULL DEFAULT 'open',
blocking_retry BOOLEAN NOT NULL DEFAULT TRUE,
message TEXT NOT NULL DEFAULT '',
upstream_http_status INTEGER NOT NULL DEFAULT 0,
upstream_response TEXT NOT NULL DEFAULT '',
occurrences INTEGER NOT NULL DEFAULT 1,
first_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
resolved_at TIMESTAMPTZ,
resolved_by BIGINT,
resolution_action VARCHAR(32) NOT NULL DEFAULT '',
resolution_note TEXT NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_rock_delivery_incidents_filter
ON rock_delivery_incidents (status, category, last_seen_at DESC);
CREATE INDEX IF NOT EXISTS idx_rock_delivery_incidents_token
ON rock_delivery_incidents (token_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_rock_delivery_incidents_order
ON rock_delivery_incidents (order_no, created_at DESC);
CREATE UNIQUE INDEX IF NOT EXISTS idx_rock_delivery_incidents_open_unique
ON rock_delivery_incidents (token_id, order_no, category)
WHERE status = 'open';
+123 -4
View File
@@ -8,6 +8,7 @@ import (
"strings"
"affiliate_dash/internal/middleware"
"affiliate_dash/internal/model"
"affiliate_dash/internal/pkg/response"
"affiliate_dash/internal/service"
@@ -15,11 +16,12 @@ import (
)
type RockHandler struct {
rockSvc *service.RockService
rockSvc *service.RockService
deliverySvc *service.DeliveryService
}
func NewRockHandler(rockSvc *service.RockService) *RockHandler {
return &RockHandler{rockSvc: rockSvc}
func NewRockHandler(rockSvc *service.RockService, deliverySvc *service.DeliveryService) *RockHandler {
return &RockHandler{rockSvc: rockSvc, deliverySvc: deliverySvc}
}
// actorPtr 把 0 值用户 ID 转成 nil,便于审计字段留空。
@@ -493,6 +495,123 @@ func (h *RockHandler) ListDeliveryLogs(c *gin.Context) {
response.OK(c, gin.H{"list": logs})
}
func (h *RockHandler) ListTokenIncidents(c *gin.Context) {
idUint, err := strconv.ParseUint(c.Param("id"), 10, 64)
if err != nil {
response.BadRequest(c, "无效的 Token ID")
return
}
list, err := h.rockSvc.ListRockTokenIncidents(c.Request.Context(), uint(idUint))
if err != nil {
response.BadRequest(c, err.Error())
return
}
response.OK(c, gin.H{"list": list})
}
func (h *RockHandler) ListIncidents(c *gin.Context) {
page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
pageSize, _ := strconv.Atoi(c.DefaultQuery("page_size", "20"))
result, err := h.rockSvc.ListRockDeliveryIncidents(
c.Request.Context(),
strings.TrimSpace(c.Query("status")),
strings.TrimSpace(c.Query("category")),
page,
pageSize,
)
if err != nil {
response.ServerError(c, err.Error())
return
}
response.OK(c, gin.H{
"list": result.List,
"total": result.Total,
"page": page,
"page_size": pageSize,
})
}
type resolveIncidentReq struct {
Note string `json:"note"`
}
func (h *RockHandler) ResolveIncident(c *gin.Context) {
idUint, err := strconv.ParseUint(c.Param("id"), 10, 64)
if err != nil || idUint == 0 {
response.BadRequest(c, "无效的异常记录 ID")
return
}
var req resolveIncidentReq
_ = c.ShouldBindJSON(&req)
if err := h.rockSvc.ResolveRockDeliveryIncident(
c.Request.Context(),
uint(idUint),
req.Note,
actorPtr(middleware.GetUserID(c)),
); err != nil {
response.BadRequest(c, err.Error())
return
}
response.OK(c, gin.H{"message": "异常记录已标记处理"})
}
type retryIncidentReq struct {
Note string `json:"note"`
}
func (h *RockHandler) RetryIncident(c *gin.Context) {
idUint, err := strconv.ParseUint(c.Param("id"), 10, 64)
if err != nil || idUint == 0 {
response.BadRequest(c, "无效的异常记录 ID")
return
}
if h.deliverySvc == nil {
response.ServerError(c, "发货服务未初始化")
return
}
var req retryIncidentReq
_ = c.ShouldBindJSON(&req)
actor := actorPtr(middleware.GetUserID(c))
plan, err := h.rockSvc.ApproveRockIncidentRetry(c.Request.Context(), uint(idUint), req.Note, actor)
if err != nil {
response.BadRequest(c, err.Error())
return
}
result, retryErr := h.deliverySvc.SubmitForMerchant(
plan.Order.MerchantID,
0,
plan.Order.OrderNo,
plan.Token.ExpectedUID,
strconv.FormatUint(uint64(plan.Token.ID), 10),
)
if retryErr != nil {
if blocking, findErr := h.rockSvc.HasBlockingRockIncident(c.Request.Context(), plan.Order.OrderNo); findErr == nil && blocking == nil {
_, _ = h.rockSvc.RecordRockIncident(
c.Request.Context(),
&plan.Token,
plan.Order.OrderNo,
model.RockIncidentCategoryOther,
"人工重试失败: "+retryErr.Error(),
0,
"",
)
}
response.BadRequest(c, "人工重试失败: "+retryErr.Error())
return
}
if err := h.rockSvc.CompleteRockIncidentRetry(
c.Request.Context(),
plan.Token.ID,
plan.Order.OrderNo,
req.Note,
actor,
); err != nil {
response.ServerError(c, "发货已提交,但关闭异常记录失败: "+err.Error())
return
}
response.OK(c, result)
}
// ReleaseToken 把使用中的提货链接释放回可用库存;上游异常时需人工强制确认。
func (h *RockHandler) ReleaseToken(c *gin.Context) {
idUint, err := strconv.ParseUint(c.Param("id"), 10, 64)
@@ -522,7 +641,7 @@ func (h *RockHandler) ReleaseToken(c *gin.Context) {
response.OK(c, gin.H{"message": message})
}
// RefreshStaleTokens 一键刷新异常 Token(绑定错误 + 使用超时 + 绑定超时,同步执行,单次最多 20 条)。
// RefreshStaleTokens 一键刷新异常 Token(绑定错误 + 待绑定超时 + 绑定后未发货,同步执行,单次最多 20 条)。
func (h *RockHandler) RefreshStaleTokens(c *gin.Context) {
result, err := h.rockSvc.RefreshAllStaleTokens(c.Request.Context())
if err != nil {
+53
View File
@@ -31,6 +31,24 @@ const (
// RockStatusAnomaly 表示「历史流转中出现过异常」的 Token,不改变其主状态。
RockStatusAnomaly = "anomaly"
// Rock 发货异常事件。
RockIncidentStatusOpen = "open"
RockIncidentStatusResolved = "resolved"
RockIncidentStatusIgnored = "ignored"
RockIncidentCategoryNoEntitlement = "no_entitlement"
RockIncidentCategoryDuplicateItem = "duplicate_item"
RockIncidentCategoryGiftExhausted = "gift_exhausted"
RockIncidentCategoryTokenRedeemed = "token_already_redeemed"
RockIncidentCategoryTokenNotRedeemable = "token_not_redeemable"
RockIncidentCategoryTokenNotFound = "token_not_found"
RockIncidentCategoryUpstreamTimeout = "upstream_timeout"
RockIncidentCategoryRateLimited = "upstream_rate_limited"
RockIncidentCategoryShipTimeout = "ship_timeout"
RockIncidentCategoryBindMismatch = "bind_mismatch"
RockIncidentCategoryOrderCancelled = "order_cancelled"
RockIncidentCategoryOther = "other"
RockSKUBlackTshirt = "rock_black_tshirt"
RockSKUBlueSportswear = "rock_blue_sportswear"
@@ -137,6 +155,13 @@ type RockToken struct {
AnomalyReason string `gorm:"-" json:"anomaly_reason,omitempty"`
AnomalyAt *time.Time `gorm:"-" json:"anomaly_at,omitempty"`
AnomalyCount int `gorm:"-" json:"anomaly_count"`
// 当前开放异常,供列表筛选和人工处理入口使用。
HasOpenIncident bool `gorm:"-" json:"has_open_incident"`
IncidentCategory string `gorm:"-" json:"incident_category,omitempty"`
IncidentMessage string `gorm:"-" json:"incident_message,omitempty"`
IncidentAt *time.Time `gorm:"-" json:"incident_at,omitempty"`
IncidentCount int64 `gorm:"-" json:"incident_count,omitempty"`
}
func (RockToken) TableName() string {
@@ -193,6 +218,34 @@ func (RockDeliveryLog) TableName() string {
return "rock_delivery_logs"
}
// RockDeliveryIncident 是需要人工处理的发货异常事件。
// 原始报文仍保存在 rock_delivery_logs,本表只保存分类和处理状态。
type RockDeliveryIncident struct {
ID uint `gorm:"primaryKey" json:"id"`
TokenID uint `gorm:"not null;index:idx_rock_delivery_incidents_token" json:"token_id"`
OrderNo string `gorm:"size:128;not null;default:'';index:idx_rock_delivery_incidents_order" json:"order_no"`
UpstreamTokenID string `gorm:"size:64;not null;default:''" json:"upstream_token_id"`
Category string `gorm:"size:40;not null;index:idx_rock_delivery_incidents_filter" json:"category"`
Status string `gorm:"size:16;not null;default:'open';index:idx_rock_delivery_incidents_filter" json:"status"`
BlockingRetry bool `gorm:"not null;default:true" json:"blocking_retry"`
Message string `gorm:"type:text;not null;default:''" json:"message"`
UpstreamHTTPStatus int `gorm:"not null;default:0" json:"upstream_http_status"`
UpstreamResponse string `gorm:"type:text;not null;default:''" json:"upstream_response"`
Occurrences int `gorm:"not null;default:1" json:"occurrences"`
FirstSeenAt time.Time `json:"first_seen_at"`
LastSeenAt time.Time `json:"last_seen_at"`
ResolvedAt *time.Time `json:"resolved_at,omitempty"`
ResolvedBy *uint `gorm:"index" json:"resolved_by,omitempty"`
ResolutionAction string `gorm:"size:32;not null;default:''" json:"resolution_action"`
ResolutionNote string `gorm:"type:text;not null;default:''" json:"resolution_note"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
func (RockDeliveryIncident) TableName() string {
return "rock_delivery_incidents"
}
// RockPurchaseLog 是一次采购尝试的主记录。
// 每一次真正进入采购流程的尝试都会落一条,包含降级原因与各阶段汇总。
type RockPurchaseLog struct {
+4
View File
@@ -301,6 +301,10 @@ func Setup(h *Handlers) *gin.Engine {
platform.POST("/platform/rock/tokens/:id/release-check", h.Rock.CheckReleaseToken)
platform.GET("/platform/rock/tokens/:id/status-logs", h.Rock.ListTokenStatusLogs)
platform.GET("/platform/rock/tokens/:id/delivery-logs", h.Rock.ListDeliveryLogs)
platform.GET("/platform/rock/tokens/:id/incidents", h.Rock.ListTokenIncidents)
platform.GET("/platform/rock/incidents", h.Rock.ListIncidents)
platform.PATCH("/platform/rock/incidents/:id", h.Rock.ResolveIncident)
platform.POST("/platform/rock/incidents/:id/retry", h.Rock.RetryIncident)
platform.POST("/platform/rock/tokens/:id/refresh-binding", h.Rock.RefreshBinding)
platform.POST("/platform/rock/token-batches/refresh-stale", h.Rock.RefreshStaleTokens)
platform.GET("/platform/rock/token-batches/preview", h.Rock.PreviewTokenDelete)
+1 -1
View File
@@ -54,7 +54,7 @@ func newTestRouterWithDB(t *testing.T) (*gin.Engine, *jwt.Manager, *gorm.DB) {
SourceOpen: handler.NewOpenHandler(fulfillmentSvc),
Merchant: handler.NewMerchantHandler(merchantSvc, fulfillmentSvc, callbackSvc, deliverySvc, rechargeSvc),
Recharge: handler.NewRechargeHandler(rechargeSvc),
Rock: handler.NewRockHandler(rockSvc),
Rock: handler.NewRockHandler(rockSvc, deliverySvc),
JWT: jm,
Tenant: tenantSvc,
OpenDB: db,
+13
View File
@@ -32,6 +32,9 @@ func deliverySubmittedUpstream(order *model.FulfillmentOrder) bool {
if order == nil {
return false
}
if deliveryPendingUpstream(order) {
return true
}
if strings.TrimSpace(order.ProviderOrderNo) != "" {
return true
}
@@ -45,6 +48,16 @@ func deliverySubmittedUpstream(order *model.FulfillmentOrder) bool {
return resultDataString(order.ResultData, "provider_order_stage") == deliveryStageSubmitted
}
// deliveryPendingUpstream 表示 Rock 已受理兑换,正在等待上游确认。
// pending 不等于“提交中断”,订单超时扫描必须跳过。
func deliveryPendingUpstream(order *model.FulfillmentOrder) bool {
if order == nil {
return false
}
stage := resultDataString(order.ResultData, "provider_order_stage")
return stage == "redeem_pending" || boolFromMap(resultDataMap(order.ResultData), "rock_redeem_pending")
}
type DeliveryService struct {
fulfillment *FulfillmentService
bffBaseURL string
@@ -65,6 +65,10 @@ func (s *FulfillmentService) markProcessingTimeout(id uint, timeout time.Duratio
if !processingTimedOut(&order, timeout, now) {
return nil
}
// Rock pending 已经提交上游,由 Rock 确认任务继续轮询,不能按“提交中断”标记失败。
if deliveryPendingUpstream(&order) {
return nil
}
if err := validateOrderStatusTransition(&order, model.OrderStatusShipFailed, fulfillmentTransitionTimeout); err != nil {
return nil
}
+105 -23
View File
@@ -6,8 +6,11 @@ import (
"fmt"
"strconv"
"strings"
"time"
"affiliate_dash/internal/model"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
// RockProvider 实现 Rock 发货平台履约逻辑
@@ -56,6 +59,17 @@ func (p *RockProvider) PrepareDelivery(ctx context.Context, order *model.Fulfill
}
if orderNo != "" {
incident, incidentErr := p.rockService.HasBlockingRockIncident(ctx, orderNo)
if incidentErr != nil {
return nil, incidentErr
}
if incident != nil {
return &DeliveryOrderState{
CanShip: false,
CannotShipReason: "该订单存在未处理发货异常,请先在 Rock 发货记录中处理",
Provider: ProviderRock,
}, nil
}
var assigned model.RockToken
findErr := p.rockService.db.WithContext(ctx).Unscoped().
Where("assigned_order_no = ? AND status IN ?", orderNo, []string{
@@ -256,6 +270,12 @@ func (p *RockProvider) Submit(ctx context.Context, order *model.FulfillmentOrder
"provider": ProviderRock,
})
if incident, incidentErr := p.rockService.HasBlockingRockIncident(ctx, order.OrderNo); incidentErr != nil {
return nil, incidentErr
} else if incident != nil {
return nil, fmt.Errorf("该订单存在未处理发货异常(%s),请人工处理后再重试", incident.Category)
}
result, err := p.rockService.RedeemTokenForOrder(ctx, order.OrderNo, sku, expectedUID, recorder)
if err != nil {
recorder.Internal(ctx, "order_update", "failed", err.Error(), map[string]interface{}{
@@ -404,32 +424,94 @@ func (p *RockProvider) Cancel(ctx context.Context, order *model.FulfillmentOrder
return nil
}
// 若订单尚未扫码绑定成功,将处于 using 状态的 Token 安全重置回 available
var token model.RockToken
err := p.rockService.db.WithContext(ctx).Unscoped().
Where("assigned_order_no = ? AND status = ?", order.OrderNo, model.RockStatusUsing).
err := p.rockService.db.WithContext(ctx).
Where("assigned_order_no = ? AND status IN ?", order.OrderNo,
[]string{model.RockStatusUsing, model.RockStatusBound, model.RockStatusRedeemed}).
Order("id DESC").
First(&token).Error
if err == nil {
fromStatus := token.Status
if updateErr := p.rockService.db.WithContext(ctx).Unscoped().Model(&token).Updates(map[string]interface{}{
"status": model.RockStatusAvailable,
"assigned_order_no": "",
"expected_uid": "",
"actual_role_name": "",
"actual_role_id": "",
"actual_channel": "",
"initial_bound_snapshot": "",
"mismatch_at": nil,
"mismatch_reason": "",
"bound_at": nil,
"using_at": nil,
"redeemed_at": nil,
"upstream_fulfill_id": 0,
"restocked_at": nil, // 订单取消回滚不属于「释放回库存」,不产生优先权
}).Error; updateErr == nil {
writeRockTokenStatusChange(p.rockService.db.WithContext(ctx), &token, fromStatus, model.RockStatusAvailable,
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
if err != nil {
return err
}
switch token.Status {
case model.RockStatusUsing:
// 尚未扫码绑定,可以安全撤销本轮领号并回库存。
retryAsBound := false
err := p.rockService.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&token, token.ID).Error; err != nil {
return err
}
if token.Status != model.RockStatusUsing {
retryAsBound = token.Status == model.RockStatusBound ||
token.Status == model.RockStatusRedeemed
return nil
}
fromStatus := token.Status
now := time.Now()
if err := tx.Unscoped().Model(&token).Updates(map[string]interface{}{
"status": model.RockStatusAvailable,
"assigned_order_no": "",
"expected_uid": "",
"actual_role_name": "",
"actual_role_id": "",
"actual_channel": "",
"initial_bound_snapshot": "",
"mismatch_at": nil,
"mismatch_reason": "",
"bound_at": nil,
"using_at": nil,
"redeemed_at": nil,
"upstream_fulfill_id": 0,
"redeem_requested_at": nil,
"redeem_checked_at": nil,
"redeem_attempts": 0,
"redeem_next_retry_at": nil,
"redeem_last_message": "",
"redeem_last_error": "",
"redeem_timeout_warned_at": nil,
"restocked_at": nil, // 订单取消回滚不属于「释放回库存」,不产生优先权
"updated_at": now,
}).Error; err != nil {
return err
}
writeRockTokenStatusChange(tx, &token, fromStatus, model.RockStatusAvailable,
rockTokenSourceOrderCancel, order.OrderNo, false, nil)
return resolveOpenRockIncidents(tx, token.ID, "order_cancel", order.OrderNo, nil)
})
if err != nil {
return err
}
if retryAsBound {
return p.Cancel(ctx, order)
}
return nil
case model.RockStatusBound:
_, err = p.rockService.RecordRockIncident(
ctx,
&token,
order.OrderNo,
model.RockIncidentCategoryOrderCancelled,
"订单已取消,但 Token 已扫码绑定;禁止自动释放,请人工核对后释放或上报失效",
0,
"",
)
return err
case model.RockStatusRedeemed:
_, err = p.rockService.RecordRockIncident(
ctx,
&token,
order.OrderNo,
model.RockIncidentCategoryOrderCancelled,
"订单已取消,但上游已发货;需要人工核对退款与履约",
0,
"",
)
return err
default:
return nil
}
return nil
}
+13 -5
View File
@@ -215,7 +215,8 @@ func TestRockProvider_AntiGhostAndUIDMismatch(t *testing.T) {
// 3. 买家扫错了号!拿了微信小号扫码,实际角色变成了“错误小号”(9999999999)
currentBoundAccount = "错误小号"
currentBoundRoleID = "9999999999"
currentBindChangedAt = "2026-10-05 14:00:00"
currentBindChangedAt = time.Now().In(timeutil.Location()).Add(time.Minute).Format("2006-01-02 15:04:05")
rockSvc.refreshCache.Delete("151153")
check2, err := provider.CheckBind(context.Background(), order, bindRes.BindUUID)
if err != nil {
@@ -275,12 +276,14 @@ func TestRockProvider_AntiGhostAndUIDMismatch(t *testing.T) {
// 5. 重新绑定正确账号“感性的洁少”(3688825401)
currentBoundAccount = "感性的洁少"
currentBoundRoleID = "3688825401"
currentBindChangedAt = "2026-10-05 14:05:00"
currentBindChangedAt = ""
bindRes2, err := provider.Bind(context.Background(), order, expectedUID)
if err != nil {
t.Fatalf("rebind error: %v", err)
}
currentBindChangedAt = time.Now().In(timeutil.Location()).Add(time.Minute).Format("2006-01-02 15:04:05")
rockSvc.refreshCache.Delete("151153")
check3, err := provider.CheckBind(context.Background(), order, bindRes2.BindUUID)
if err != nil {
t.Fatalf("check bind match: %v", err)
@@ -638,6 +641,7 @@ func TestReleaseUnboundUsingToken(t *testing.T) {
}
// 上游已经出现新扫码时拒绝释放。
releaseBaseline := time.Now().Add(-time.Hour)
using := model.RockToken{
UpstreamTokenID: "88002",
Token: "jwt2",
@@ -646,12 +650,13 @@ func TestReleaseUnboundUsingToken(t *testing.T) {
Status: model.RockStatusUsing,
AssignedOrderNo: "FO-RELEASE-2",
ExpectedUID: "3688825401",
ClaimedAt: &releaseBaseline,
}
if err := db.Create(&using).Error; err != nil {
t.Fatalf("create using token: %v", err)
}
boundRoleID = "3688825401"
bindChangedAt = "2026-10-05 21:10:00"
bindChangedAt = time.Now().In(timeutil.Location()).Add(-30 * time.Minute).Format("2006-01-02 15:04:05")
if err := rockSvc.ReleaseUnboundUsingToken(context.Background(), using.ID, false, nil); err == nil || !strings.Contains(err.Error(), "已扫码绑定") {
t.Fatalf("已绑定链接必须拒绝释放,实际: %v", err)
}
@@ -705,13 +710,15 @@ func TestReleaseUnboundUsingToken(t *testing.T) {
boundRoleID = "4744920014"
bindChangedAt = "2026-10-05 21:52:30"
deliveryTime = "2026-10-05T21:52:40.61"
if err := rockSvc.ReleaseUnboundUsingToken(context.Background(), fresh.ID, false, nil); err == nil || !strings.Contains(err.Error(), "已扫码绑定") {
if err := rockSvc.ReleaseUnboundUsingToken(context.Background(), fresh.ID, false, nil); err == nil ||
(!strings.Contains(err.Error(), "已扫码绑定") && !strings.Contains(err.Error(), "上游已发货")) {
t.Fatalf("领取后的新发货必须拒绝释放,实际: %v", err)
}
}
func TestRockProvider_SchemeB_UIDSelfHealing(t *testing.T) {
db := newRockTestDB(t)
currentBindChangedAt := ""
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
@@ -728,7 +735,7 @@ func TestRockProvider_SchemeB_UIDSelfHealing(t *testing.T) {
"token": map[string]interface{}{
"id": 152001,
"status": "bound",
"bind_changed_at": "2026-10-06 10:00:00",
"bind_changed_at": currentBindChangedAt,
"bound_account": "少年有点忧郁",
"bound_account_id": "3454648872",
"region": "安卓-微信",
@@ -771,6 +778,7 @@ func TestRockProvider_SchemeB_UIDSelfHealing(t *testing.T) {
if bindRes1.Bound {
t.Fatalf("尚未扫码,bound 应为 false")
}
currentBindChangedAt = time.Now().In(timeutil.Location()).Add(time.Minute).Format("2006-01-02 15:04:05")
// 2. 扫码反查,真实角色是 3454648872,发生 mismatch
check1, err := provider.CheckBind(ctx, order, bindRes1.BindUUID)
+24 -13
View File
@@ -20,13 +20,16 @@ import (
//
// 采用子查询而不是落库字段,保证历史数据无需迁移即可被筛选到。
const rockAnomalyLogConditionSQL = `(
l.to_status IN ('mismatch', 'abnormal')
OR l.reason LIKE '%失败%'
OR l.reason LIKE '%不存在%'
OR l.reason LIKE '%异常%'
OR l.reason LIKE '%超时%'
OR l.reason LIKE '%拦截%'
OR l.reason LIKE '%不符%'
l.event_type <> 'incident'
AND (
l.to_status IN ('mismatch', 'abnormal')
OR l.reason LIKE '%失败%'
OR l.reason LIKE '%不存在%'
OR l.reason LIKE '%异常%'
OR l.reason LIKE '%超时%'
OR l.reason LIKE '%拦截%'
OR l.reason LIKE '%不符%'
)
)`
// rockAnomalyTokenScopeSQL 排除已有明确结论的 Token(别名固定为 t):
@@ -34,12 +37,20 @@ const rockAnomalyLogConditionSQL = `(
const rockAnomalyTokenScopeSQL = `t.status NOT IN ('redeemed', 'abnormal') AND t.redeemed_at IS NULL`
func rockTokenAnomalyExistsSQL() string {
return `EXISTS (
SELECT 1 FROM rock_token_status_logs l
JOIN rock_tokens t ON t.id = l.token_id
WHERE l.token_id = rock_tokens.id
AND ` + rockAnomalyTokenScopeSQL + `
AND ` + rockAnomalyLogConditionSQL + `
return `(
EXISTS (
SELECT 1 FROM rock_token_status_logs l
JOIN rock_tokens t ON t.id = l.token_id
WHERE l.token_id = rock_tokens.id
AND ` + rockAnomalyTokenScopeSQL + `
AND ` + rockAnomalyLogConditionSQL + `
)
OR EXISTS (
SELECT 1 FROM rock_delivery_incidents i
WHERE i.token_id = rock_tokens.id
AND i.status = 'open'
AND i.category <> 'no_entitlement'
)
)`
}
+15 -10
View File
@@ -16,11 +16,13 @@ import (
// BindingSnapshot 保存分配 Token 时上游的绑定快照
type BindingSnapshot struct {
Bound bool `json:"bound"`
RoleName string `json:"role_name"`
RoleID string `json:"role_id"`
Channel string `json:"channel"`
BindChangedAt string `json:"bind_changed_at"`
Bound bool `json:"bound"`
RoleName string `json:"role_name"`
RoleID string `json:"role_id"`
Channel string `json:"channel"`
BindChangedAt string `json:"bind_changed_at"`
BindChangeType string `json:"bind_change_type,omitempty"`
Status string `json:"status,omitempty"`
}
// rockOrderTokenLockedMessage 一单只能占用一个提货号。绑定失败后不再发新号。
@@ -166,12 +168,15 @@ func (s *RockService) AssignTokenForOrder(ctx context.Context, orderNo, expected
if token.InitialBoundSnapshot == "" && cfg.BearerToken != "" {
ref, refErr := s.client.RefreshToken(ctx, cfg.BearerToken, token.UpstreamTokenID)
if refErr == nil && ref != nil {
currentBound, _ := rockCurrentCycleBinding(&token, ref)
snap := BindingSnapshot{
Bound: ref.IsBound(),
RoleName: ref.GetRoleName(),
RoleID: ref.GetRoleID(),
Channel: ref.GetChannel(),
BindChangedAt: ref.Token.BindChangedAt,
Bound: currentBound,
RoleName: ref.GetRoleName(),
RoleID: ref.GetRoleID(),
Channel: ref.GetChannel(),
BindChangedAt: ref.Token.BindChangedAt,
BindChangeType: ref.Token.BindChangeType,
Status: ref.Token.Status,
}
snapBytes, _ := json.Marshal(snap)
token.InitialBoundSnapshot = string(snapBytes)
+76 -32
View File
@@ -7,6 +7,7 @@ import (
"time"
"affiliate_dash/internal/model"
"gorm.io/gorm"
)
type CheckBindingResult struct {
@@ -46,16 +47,27 @@ func (s *RockService) CheckBindingAndVerifyUID(ctx context.Context, tokenID uint
token.MismatchReason = ""
token.MismatchAt = nil
token.BoundAt = &now
_ = s.db.WithContext(ctx).Unscoped().Model(&token).Updates(map[string]interface{}{
"expected_uid": trimmedExpected,
"status": model.RockStatusBound,
"mismatch_reason": "",
"mismatch_at": nil,
"bound_at": now,
"updated_at": now,
}).Error
writeRockTokenStatusChange(s.db.WithContext(ctx), &token, model.RockStatusMismatch, model.RockStatusBound,
rockTokenSourceBindCheck, "买家确认正确 UID 后解除绑定错误", false, nil)
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Unscoped().Model(&token).Updates(map[string]interface{}{
"expected_uid": trimmedExpected,
"status": model.RockStatusBound,
"mismatch_reason": "",
"mismatch_at": nil,
"bound_at": now,
"updated_at": now,
}).Error; err != nil {
return err
}
writeRockTokenStatusChange(tx, &token, model.RockStatusMismatch, model.RockStatusBound,
rockTokenSourceBindCheck, "买家确认正确 UID 后解除绑定错误", false, nil)
return writeAudit(tx, nil, nil, nil, "rock_token.bind_confirm", "rock_token",
fmt.Sprintf("%d", token.ID), map[string]interface{}{
"expected_uid": trimmedExpected,
"actual_uid": actualUID,
})
}); err != nil {
return nil, fmt.Errorf("更新扫码绑定状态失败: %w", err)
}
return &CheckBindingResult{
Bound: true,
RoleName: token.ActualRoleName,
@@ -155,7 +167,7 @@ func (s *RockService) CheckBindingAndVerifyUID(ctx context.Context, tokenID uint
}
// 1. 检查上游是否已绑定 (上游状态 bound 或 bound_account_id 有值)
if !ref.IsBound() || strings.TrimSpace(ref.GetRoleID()) == "" {
if !ref.HasBindingSignal() || strings.TrimSpace(ref.GetRoleID()) == "" {
return &CheckBindingResult{
Bound: false,
Message: "等待买家微信/QQ扫码授权",
@@ -165,9 +177,10 @@ func (s *RockService) CheckBindingAndVerifyUID(ctx context.Context, tokenID uint
actualRoleName := ref.GetRoleName()
actualRoleID := ref.GetRoleID()
actualChannel := ref.GetChannel()
currentBound, bindChangedAt := rockCurrentCycleBinding(&token, ref)
// 2. 历史残留不算本次扫码:角色和绑定时间与分配快照一致,或发货记录都早于本次领取。
if !rockBuyerNewlyBound(&token, ref) {
// 2. 历史残留不算本次扫码:只认晚于本轮 using_at 的 bind_changed_at。
if !currentBound {
return &CheckBindingResult{
Bound: false,
Message: "等待买家扫码(检测到历史绑定残留,已自动忽略)",
@@ -185,16 +198,30 @@ func (s *RockService) CheckBindingAndVerifyUID(ctx context.Context, tokenID uint
// 严禁发货!跃迁为 mismatch
fromStatus := token.Status
mismatchReason := fmt.Sprintf("预期UID(%s) 与 扫码UID(%s/%s) 不一致", trimmedExpected, actualRoleName, actualRoleID)
_ = s.db.WithContext(ctx).Unscoped().Model(&token).Updates(map[string]interface{}{
"status": model.RockStatusMismatch,
"actual_role_name": actualRoleName,
"actual_role_id": actualRoleID,
"actual_channel": actualChannel,
"mismatch_at": now,
"mismatch_reason": mismatchReason,
}).Error
writeRockTokenStatusChange(s.db.WithContext(ctx), &token, fromStatus, model.RockStatusMismatch,
rockTokenSourceBindCheck, mismatchReason, false, nil)
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Unscoped().Model(&token).Updates(map[string]interface{}{
"status": model.RockStatusMismatch,
"actual_role_name": actualRoleName,
"actual_role_id": actualRoleID,
"actual_channel": actualChannel,
"mismatch_at": now,
"mismatch_reason": mismatchReason,
"bound_at": bindChangedAt,
"updated_at": now,
}).Error; err != nil {
return err
}
writeRockTokenStatusChange(tx, &token, fromStatus, model.RockStatusMismatch,
rockTokenSourceBindCheck, mismatchReason, false, nil)
return writeAudit(tx, nil, nil, nil, "rock_token.bind_mismatch", "rock_token",
fmt.Sprintf("%d", token.ID), map[string]interface{}{
"expected_uid": trimmedExpected,
"actual_uid": actualRoleID,
"reason": mismatchReason,
})
}); err != nil {
return nil, fmt.Errorf("记录绑定 UID 不一致失败: %w", err)
}
roleDesc := actualRoleName
if roleDesc == "" {
@@ -213,15 +240,32 @@ func (s *RockService) CheckBindingAndVerifyUID(ctx context.Context, tokenID uint
// 4. UID 一致,更新为 bound 成功状态并保存玩家角色名称
fromStatus := token.Status
_ = s.db.WithContext(ctx).Unscoped().Model(&token).Updates(map[string]interface{}{
"status": model.RockStatusBound,
"actual_role_name": actualRoleName,
"actual_role_id": actualRoleID,
"actual_channel": actualChannel,
"bound_at": now,
}).Error
writeRockTokenStatusChange(s.db.WithContext(ctx), &token, fromStatus, model.RockStatusBound,
rockTokenSourceBindCheck, fmt.Sprintf("扫码绑定成功 UID %s", actualRoleID), false, nil)
boundAt := bindChangedAt
if boundAt.IsZero() {
boundAt = now
}
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Unscoped().Model(&token).Updates(map[string]interface{}{
"status": model.RockStatusBound,
"actual_role_name": actualRoleName,
"actual_role_id": actualRoleID,
"actual_channel": actualChannel,
"bound_at": boundAt,
"updated_at": now,
}).Error; err != nil {
return err
}
writeRockTokenStatusChange(tx, &token, fromStatus, model.RockStatusBound,
rockTokenSourceBindCheck, fmt.Sprintf("扫码绑定成功 UID %s", actualRoleID), false, nil)
return writeAudit(tx, nil, nil, nil, "rock_token.bind_confirm", "rock_token",
fmt.Sprintf("%d", token.ID), map[string]interface{}{
"expected_uid": trimmedExpected,
"actual_uid": actualRoleID,
"bound_at": boundAt.Format(time.RFC3339),
})
}); err != nil {
return nil, fmt.Errorf("保存扫码绑定状态失败: %w", err)
}
return &CheckBindingResult{
Bound: true,
@@ -36,12 +36,12 @@ func canRefreshRockBinding(status string) bool {
// RefreshTokenBinding 通过兑换码调用上游 refresh,人工同步一次绑定状态。
//
// 规则(只同步上游实时状态,不再用本地预期 UID 拦截):
// 规则(只同步本轮真实事件,并与本地预期 UID 交叉校验):
// - 上游状态异常(abnormal)→ 本地同步 abnormal(已上报失效);
// - 上游明确返回「Token 不存在」(404) → 本地同步 abnormal(已上报失效);
// - 上游已发货 → 本地同步 redeemed;
// - 上游已绑定 → 本地直接同步 bound,并清空绑定错误标记;
// - 上游未绑定 → 本地状态保持不变(不自动把 bound 退回 using)。
// - 本轮 redeemed_at / deliveryList → 本地同步 redeemed;
// - 本轮 bind_changed_at → 本地同步 bound/mismatch;
// - 只有历史残留事件 → 不推进本轮状态,错误的 bound 修复回 using。
func (s *RockService) RefreshTokenBinding(ctx context.Context, tokenID uint) (*TokenBindingRefreshResult, error) {
return s.refreshTokenBinding(ctx, tokenID, nil)
}
@@ -119,13 +119,13 @@ func (s *RockService) refreshTokenBinding(ctx context.Context, tokenID uint, act
actualRoleName := ref.GetRoleName()
actualRoleID := ref.GetRoleID()
actualChannel := ref.GetChannel()
upstreamBound := rockUpstreamBound(ref)
currentBound, bindChangedAt := rockCurrentCycleBinding(&token, ref)
updates := map[string]interface{}{
"updated_at": now,
}
// 只有上游确实绑定时才同步角色字段;未绑定时 Region 可能只是领取区域,不能当成绑定渠道。
if upstreamBound {
if currentBound {
if actualRoleName != "" {
updates["actual_role_name"] = actualRoleName
}
@@ -139,37 +139,42 @@ func (s *RockService) refreshTokenBinding(ctx context.Context, tokenID uint, act
result := &TokenBindingRefreshResult{
Status: token.Status,
Bound: upstreamBound,
Bound: currentBound,
}
if upstreamBound {
if currentBound {
result.RoleName = actualRoleName
result.RoleID = actualRoleID
result.Channel = actualChannel
}
redeemedAtRaw := strings.TrimSpace(ref.Token.RedeemedAt)
deliveryEvidence := rockCurrentCycleDelivery(&token, ref)
baselineUnknown := rockClaimBaseline(&token).IsZero()
matchesInitialSnapshot := rockBindingMatchesInitialSnapshot(&token, ref)
switch {
case upstreamStatus == model.RockStatusRedeemed || redeemedAtRaw != "":
redeemedAt, ok := parseRockTime(redeemedAtRaw)
if !ok {
case deliveryEvidence.Ambiguous:
result.Warning = true
result.Message = "上游显示已发货,但无法确认是否属于本轮,禁止释放,请人工核对"
case deliveryEvidence.Current:
redeemedAt := deliveryEvidence.At
if redeemedAt.IsZero() {
redeemedAt = now
}
updates["status"] = model.RockStatusRedeemed
updates["redeemed_at"] = redeemedAt
if len(ref.DeliveryList) > 0 {
if fulfillID := ref.DeliveryList[len(ref.DeliveryList)-1].FulfillID; fulfillID > 0 {
updates["upstream_fulfill_id"] = fulfillID
}
if deliveryEvidence.FulfillID > 0 {
updates["upstream_fulfill_id"] = deliveryEvidence.FulfillID
}
result.Status = model.RockStatusRedeemed
result.RedeemedAt = redeemedAt.Format("2006-01-02 15:04:05")
result.Message = "上游已发货,本地状态已同步为已交付"
case upstreamBound:
// 上游已绑定即同步为 bound:不以本地预期 UID 做拦截,
// 否则历史 UID 不一致的 Token 永远无法把真实状态刷新回本地。
// 发货确认中(有 redeem 标记未完成)时只刷新标记,状态保持 bound。
case baselineUnknown && !matchesInitialSnapshot &&
(ref.HasBindingSignal() || strings.TrimSpace(rockBindingChangedAt(ref)) != ""):
result.Warning = true
result.Message = "缺少本轮 using_at,无法判断该绑定是否属于本轮,请人工核对"
case currentBound:
if rockTokenRedeemPending(&token) {
updates["redeem_checked_at"] = now
if msg := strings.TrimSpace(ref.Message); msg != "" {
@@ -179,12 +184,21 @@ func (s *RockService) refreshTokenBinding(ctx context.Context, tokenID uint, act
result.Message = "上游仍未确认发货,保持发货确认中"
break
}
updates["bound_at"] = bindChangedAt
expectedUID := strings.TrimSpace(token.ExpectedUID)
if expectedUID != "" && actualRoleID != "" && expectedUID != actualRoleID {
reason := fmt.Sprintf("预期UID(%s) 与 扫码UID(%s/%s) 不一致", expectedUID, actualRoleName, actualRoleID)
updates["status"] = model.RockStatusMismatch
updates["mismatch_at"] = now
updates["mismatch_reason"] = reason
result.Status = model.RockStatusMismatch
result.Warning = true
result.Message = reason
break
}
updates["status"] = model.RockStatusBound
updates["mismatch_at"] = nil
updates["mismatch_reason"] = ""
if token.BoundAt == nil {
updates["bound_at"] = resolveRockBoundAt(ref, now)
}
result.Status = model.RockStatusBound
if actualRoleID == "" {
result.Warning = true
@@ -201,18 +215,38 @@ func (s *RockService) refreshTokenBinding(ctx context.Context, tokenID uint, act
updates["redeem_checked_at"] = now
result.Status = model.RockStatusBound
result.Message = "上游仍未确认发货,保持发货确认中"
} else if token.Status == model.RockStatusBound {
updates["status"] = model.RockStatusUsing
updates["bound_at"] = nil
updates["actual_role_name"] = ""
updates["actual_role_id"] = ""
updates["actual_channel"] = ""
updates["mismatch_at"] = nil
updates["mismatch_reason"] = ""
result.Status = model.RockStatusUsing
result.Warning = true
result.Message = "未检测到本轮真实绑定事件,已修复为使用中"
} else {
result.Message = "上游仍未绑定,本地状态保持不变"
}
}
result.Changed = result.Status != originalStatus ||
(upstreamBound && actualRoleName != "" && actualRoleName != originalRoleName) ||
(upstreamBound && actualRoleID != "" && actualRoleID != originalRoleID) ||
(upstreamBound && actualChannel != "" && actualChannel != originalChannel)
(currentBound && actualRoleName != "" && actualRoleName != originalRoleName) ||
(currentBound && actualRoleID != "" && actualRoleID != originalRoleID) ||
(currentBound && actualChannel != "" && actualChannel != originalChannel)
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Model(&token).Updates(updates).Error; err != nil {
updateResult := tx.Model(&model.RockToken{}).
Where("id = ? AND updated_at = ?", token.ID, token.UpdatedAt).
Updates(updates)
if updateResult.Error != nil {
return updateResult.Error
}
if updateResult.RowsAffected == 0 {
return errors.New("Token 状态已被并发更新,请刷新后重试")
}
if err := tx.First(&token, token.ID).Error; err != nil {
return err
}
writeRockTokenStatusChange(tx, &token, originalStatus, result.Status,
@@ -308,12 +342,3 @@ func (s *RockService) markTokenAbnormalByRefresh(ctx context.Context, token *mod
Message: message,
}, nil
}
func resolveRockBoundAt(ref *RockRefreshResponse, fallback time.Time) time.Time {
if ref != nil {
if t, ok := parseRockTime(ref.Token.BindChangedAt); ok {
return t
}
}
return fallback
}
@@ -10,8 +10,13 @@ import (
"time"
"affiliate_dash/internal/model"
"affiliate_dash/internal/pkg/timeutil"
)
func recentRockBindTime(base time.Time) string {
return base.Add(time.Minute).In(timeutil.Location()).Format("2006-01-02 15:04:05")
}
func TestRefreshTokenBinding_BoundMatchesExpectedUID(t *testing.T) {
db := newRockTestDB(t)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -25,6 +30,7 @@ func TestRefreshTokenBinding_BoundMatchesExpectedUID(t *testing.T) {
"status": "bound",
"bound_account": "痛写心扉",
"bound_account_id": "3132710334",
"bind_changed_at": recentRockBindTime(time.Now()),
"region": "安卓-微信",
},
"binding": map[string]interface{}{
@@ -79,8 +85,8 @@ func TestRefreshTokenBinding_BoundMatchesExpectedUID(t *testing.T) {
}
}
// 刷新链路只同步上游实时状态:即使本地预期 UID 与扫码 UID 不一致,也应直接标记为已绑定。
func TestRefreshTokenBinding_IgnoresUIDMismatchAndMarksBound(t *testing.T) {
// 本轮真实绑定与预期 UID 不一致时必须进入 mismatch。
func TestRefreshTokenBinding_UIDMismatchMarksMismatch(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")
@@ -89,6 +95,7 @@ func TestRefreshTokenBinding_IgnoresUIDMismatchAndMarksBound(t *testing.T) {
"status": "bound",
"bound_account": "痛写心扉",
"bound_account_id": "3132710334",
"bind_changed_at": recentRockBindTime(time.Now()),
"region": "安卓-微信",
},
})
@@ -116,19 +123,19 @@ func TestRefreshTokenBinding_IgnoresUIDMismatchAndMarksBound(t *testing.T) {
if err != nil {
t.Fatalf("刷新绑定状态失败: %v", err)
}
if res.Status != model.RockStatusBound || !res.Bound {
t.Fatalf("UID 不一致时也应同步为 bound,实际 %+v", res)
if res.Status != model.RockStatusMismatch || !res.Bound {
t.Fatalf("UID 不一致时应同步为 mismatch,实际 %+v", res)
}
var saved model.RockToken
if err := db.First(&saved, token.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if saved.Status != model.RockStatusBound || saved.ActualRoleID != "3132710334" {
t.Fatalf("应同步为 bound 并写入实际 UID,实际 %+v", saved)
if saved.Status != model.RockStatusMismatch || saved.ActualRoleID != "3132710334" {
t.Fatalf("应同步为 mismatch 并写入实际 UID,实际 %+v", saved)
}
if saved.MismatchAt != nil || saved.MismatchReason != "" {
t.Fatalf("同步为 bound 后应清空绑定错误标记,实际 %+v", saved)
if saved.MismatchAt == nil || saved.MismatchReason == "" {
t.Fatalf("UID 不一致必须写入绑定错误标记,实际 %+v", saved)
}
if saved.BoundAt == nil {
t.Fatalf("bound_at 未写入")
@@ -141,6 +148,9 @@ func TestRefreshTokenBinding_EmptyExpectedUIDMarksBound(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"token": map[string]interface{}{
"bind_changed_at": recentRockBindTime(time.Now()),
},
"binding": map[string]interface{}{
"bound": true,
"role": "痛写心扉",
@@ -154,11 +164,13 @@ func TestRefreshTokenBinding_EmptyExpectedUIDMarksBound(t *testing.T) {
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
now := time.Now()
token := model.RockToken{
UpstreamTokenID: "9003",
RedeemCode: "CODE-NOUID",
Color: model.RockColorBlack,
Status: model.RockStatusUsing,
ClaimedAt: &now,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
@@ -186,6 +198,7 @@ func TestRefreshTokenBinding_EmptyExpectedUIDMarksBound(t *testing.T) {
func TestRefreshTokenBinding_SyncsRedeemed(t *testing.T) {
db := newRockTestDB(t)
now := time.Now()
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]interface{}{
@@ -193,10 +206,10 @@ func TestRefreshTokenBinding_SyncsRedeemed(t *testing.T) {
"status": "redeemed",
"bound_account": "痛写心扉",
"bound_account_id": "3132710334",
"redeemed_at": "2026-10-29 12:00:00",
"redeemed_at": now.Add(-20 * time.Second).In(timeutil.Location()).Format("2006-01-02 15:04:05"),
},
"deliveryList": []map[string]interface{}{
{"fulfillId": 4519960, "time": "2026-10-29T12:00:00.1", "role": "安卓-微信-痛写心扉"},
{"fulfillId": 4519960, "time": now.Add(-10 * time.Second).Format("2006-01-02T15:04:05.1"), "role": "安卓-微信-痛写心扉"},
},
})
}))
@@ -212,6 +225,7 @@ func TestRefreshTokenBinding_SyncsRedeemed(t *testing.T) {
Status: model.RockStatusUsing,
AssignedOrderNo: "FO-REFRESH-4",
ExpectedUID: "3132710334",
UsingAt: &now,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
@@ -234,6 +248,140 @@ func TestRefreshTokenBinding_SyncsRedeemed(t *testing.T) {
}
}
func TestRefreshTokenBinding_IgnoresHistoricalRedeemedAt(t *testing.T) {
db := newRockTestDB(t)
now := time.Now()
historical := now.Add(-48 * time.Hour)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"token": map[string]interface{}{
"status": "redeemed",
"redeemed_at": historical.In(timeutil.Location()).Format("2006-01-02 15:04:05"),
},
"deliveryList": []map[string]interface{}{
{"fulfillId": 100, "time": historical.Format("2006-01-02T15:04:05.1")},
},
})
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
token := model.RockToken{
UpstreamTokenID: "9005",
RedeemCode: "CODE-HISTORICAL-REDEEMED",
Color: model.RockColorBlack,
Status: model.RockStatusUsing,
AssignedOrderNo: "FO-REFRESH-5",
UsingAt: &now,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
res, err := svc.RefreshTokenBinding(context.Background(), token.ID)
if err != nil {
t.Fatalf("刷新绑定状态失败: %v", err)
}
if res.Status != model.RockStatusUsing || res.Bound {
t.Fatalf("历史兑换记录不能推进本轮状态,实际 %+v", res)
}
var saved model.RockToken
if err := db.First(&saved, token.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if saved.Status != model.RockStatusUsing || saved.RedeemedAt != nil {
t.Fatalf("Token 不应被历史兑换记录改成已发货: %+v", saved)
}
}
func TestRefreshTokenBinding_SnapshotEventAcceptsDifferentTimeFormat(t *testing.T) {
db := newRockTestDB(t)
now := time.Now()
boundAt := now.Add(time.Minute)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"token": map[string]interface{}{
"status": "bound",
"bound_account": "上一任角色",
"bound_account_id": "99001",
"bind_changed_at": boundAt.Format(time.RFC3339Nano),
},
})
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
token := model.RockToken{
UpstreamTokenID: "9006",
RedeemCode: "CODE-SNAPSHOT-FORMAT",
Color: model.RockColorBlack,
Status: model.RockStatusUsing,
AssignedOrderNo: "FO-REFRESH-6",
UsingAt: &now,
InitialBoundSnapshot: `{"bound":true,"role_id":"99001","bind_changed_at":"` + boundAt.In(timeutil.Location()).Format("2006-01-02 15:04:05") + `"}`,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
res, err := svc.RefreshTokenBinding(context.Background(), token.ID)
if err != nil {
t.Fatalf("刷新绑定状态失败: %v", err)
}
if res.Status != model.RockStatusUsing || res.Bound {
t.Fatalf("与初始快照相同的绑定事件必须按残留处理,实际 %+v", res)
}
}
func TestRefreshTokenBinding_MissingBaselineDoesNotDowngradeBound(t *testing.T) {
db := newRockTestDB(t)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"token": map[string]interface{}{
"status": "bound",
"bound_account": "历史角色",
"bound_account_id": "88001",
"bind_changed_at": time.Now().In(timeutil.Location()).Add(-time.Hour).Format("2006-01-02 15:04:05"),
},
})
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
token := model.RockToken{
UpstreamTokenID: "9007",
RedeemCode: "CODE-MISSING-BASELINE",
Color: model.RockColorBlack,
Status: model.RockStatusBound,
AssignedOrderNo: "FO-REFRESH-7",
ExpectedUID: "88001",
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
res, err := svc.RefreshTokenBinding(context.Background(), token.ID)
if err != nil {
t.Fatalf("刷新绑定状态失败: %v", err)
}
if res.Status != model.RockStatusBound || !res.Warning {
t.Fatalf("缺少时间基线时只能告警,不能自动降级,实际 %+v", res)
}
var saved model.RockToken
if err := db.First(&saved, token.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if saved.Status != model.RockStatusBound {
t.Fatalf("缺少时间基线时不得自动改写 Token 状态,实际 %s", saved.Status)
}
}
func TestRefreshTokenBinding_UpstreamErrorKeepsState(t *testing.T) {
db := newRockTestDB(t)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -281,6 +429,7 @@ func TestRefreshTokenBinding_FallsBackToAdminRefresh(t *testing.T) {
"status": "bound",
"bound_account": "痛写心扉",
"bound_account_id": "3132710334",
"bind_changed_at": recentRockBindTime(time.Now()),
},
})
default:
+9 -9
View File
@@ -153,13 +153,6 @@ type RockTokenData struct {
RedeemedAt string `json:"redeemed_at"`
}
func (t *RockTokenData) IsBound() bool {
if t.Status == "bound" || t.Status == "redeemed" {
return true
}
return strings.TrimSpace(t.BoundAccountID) != ""
}
func (t *RockTokenData) GetRoleName() string {
return strings.TrimSpace(t.BoundAccount)
}
@@ -214,11 +207,18 @@ type RockRefreshResponse struct {
Error string `json:"error,omitempty"`
}
func (r *RockRefreshResponse) IsBound() bool {
// HasBindingSignal 表示上游存在任何绑定痕迹,不代表该绑定属于本轮。
// 本轮判定必须使用 rockCurrentCycleBinding。
func (r *RockRefreshResponse) HasBindingSignal() bool {
if r == nil {
return false
}
if r.Token.IsBound() {
switch strings.ToLower(strings.TrimSpace(r.Token.Status)) {
case "bound", "redeemed":
return true
}
if strings.TrimSpace(r.Token.BoundAccountID) != "" ||
strings.TrimSpace(r.Token.BindChangedAt) != "" {
return true
}
return r.Binding.Bound || strings.TrimSpace(r.Binding.RoleID) != ""
+464
View File
@@ -0,0 +1,464 @@
package service
import (
"context"
"errors"
"fmt"
"strings"
"time"
"affiliate_dash/internal/model"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
type RockIncidentListResult struct {
List []model.RockDeliveryIncident `json:"list"`
Total int64 `json:"total"`
}
type RockIncidentRetryPlan struct {
Incident model.RockDeliveryIncident
Token model.RockToken
Order model.FulfillmentOrder
}
func rockIncidentRetryable(category string) bool {
switch category {
case model.RockIncidentCategoryUpstreamTimeout, model.RockIncidentCategoryRateLimited:
return true
default:
return false
}
}
func classifyRockDeliveryError(err error) (category string, httpStatus int, response string) {
if err == nil {
return model.RockIncidentCategoryOther, 0, ""
}
var statusErr *RockUpstreamStatusError
if errors.As(err, &statusErr) {
httpStatus = statusErr.StatusCode
response = statusErr.Body
}
body := strings.ToLower(strings.TrimSpace(response + " " + err.Error()))
compactBody := strings.NewReplacer(" ", "", "\n", "", "\r", "", "\t", "").Replace(body)
switch {
case strings.Contains(body, "用户不存在可以兑换的可用权益"):
return model.RockIncidentCategoryNoEntitlement, httpStatus, response
case strings.Contains(body, "同一角色无法重复兑换已兑换过的道具"):
return model.RockIncidentCategoryDuplicateItem, httpStatus, response
case strings.Contains(body, "礼包已发放完毕"):
return model.RockIncidentCategoryGiftExhausted, httpStatus, response
case strings.Contains(compactBody, "该token已兑换"):
return model.RockIncidentCategoryTokenRedeemed, httpStatus, response
case strings.Contains(compactBody, "该token暂不可兑换"):
return model.RockIncidentCategoryTokenNotRedeemable, httpStatus, response
case strings.Contains(compactBody, "token不存在"):
return model.RockIncidentCategoryTokenNotFound, httpStatus, response
case strings.Contains(body, "请求过于频繁") || strings.Contains(body, "110001"):
return model.RockIncidentCategoryRateLimited, httpStatus, response
case strings.Contains(body, "context deadline exceeded") || strings.Contains(body, "timeout"):
return model.RockIncidentCategoryUpstreamTimeout, httpStatus, response
case strings.Contains(body, "发货超时"):
return model.RockIncidentCategoryShipTimeout, httpStatus, response
case strings.Contains(body, "uid") || strings.Contains(body, "玩家编号") || strings.Contains(body, "绑定账号"):
return model.RockIncidentCategoryBindMismatch, httpStatus, response
default:
return model.RockIncidentCategoryOther, httpStatus, response
}
}
// RecordRockDeliveryIncident 持久化一次需要人工处理的发货异常,并同步 Token 错误摘要。
func (s *RockService) RecordRockDeliveryIncident(ctx context.Context, token *model.RockToken, orderNo string, cause error) (*model.RockDeliveryIncident, error) {
if token == nil || token.ID == 0 {
return nil, errors.New("Token 不存在")
}
category, httpStatus, response := classifyRockDeliveryError(cause)
message := ""
if cause != nil {
message = strings.TrimSpace(cause.Error())
}
if message == "" {
message = "上游发货失败"
}
return s.RecordRockIncident(ctx, token, orderNo, category, message, httpStatus, response)
}
func (s *RockService) RecordRockIncident(
ctx context.Context,
token *model.RockToken,
orderNo, category, message string,
httpStatus int,
upstreamResponse string,
) (*model.RockDeliveryIncident, error) {
if token == nil || token.ID == 0 {
return nil, errors.New("Token 不存在")
}
category = strings.TrimSpace(category)
if category == "" {
category = model.RockIncidentCategoryOther
}
orderNo = strings.TrimSpace(orderNo)
message = strings.TrimSpace(message)
upstreamResponse = strings.TrimSpace(upstreamResponse)
now := time.Now()
var incident model.RockDeliveryIncident
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var current model.RockToken
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&current, token.ID).Error; err != nil {
return err
}
err := tx.Where(
"token_id = ? AND order_no = ? AND category = ? AND status = ?",
token.ID, orderNo, category, model.RockIncidentStatusOpen,
).First(&incident).Error
switch {
case err == nil:
if err := tx.Model(&incident).Updates(map[string]interface{}{
"occurrences": incident.Occurrences + 1,
"blocking_retry": true,
"message": message,
"upstream_http_status": httpStatus,
"upstream_response": upstreamResponse,
"last_seen_at": now,
"updated_at": now,
}).Error; err != nil {
return err
}
incident.Occurrences++
incident.BlockingRetry = true
incident.Message = message
incident.UpstreamHTTPStatus = httpStatus
incident.UpstreamResponse = upstreamResponse
incident.LastSeenAt = now
case errors.Is(err, gorm.ErrRecordNotFound):
incident = model.RockDeliveryIncident{
TokenID: current.ID,
OrderNo: orderNo,
UpstreamTokenID: strings.TrimSpace(current.UpstreamTokenID),
Category: category,
Status: model.RockIncidentStatusOpen,
BlockingRetry: true,
Message: message,
UpstreamHTTPStatus: httpStatus,
UpstreamResponse: upstreamResponse,
Occurrences: 1,
FirstSeenAt: now,
LastSeenAt: now,
}
if err := tx.Create(&incident).Error; err != nil {
return err
}
default:
return err
}
if err := tx.Model(&current).Updates(map[string]interface{}{
"redeem_last_error": message,
"remark": appendRockRemark(current.Remark, message),
"updated_at": now,
}).Error; err != nil {
return err
}
writeRockTokenStatusLog(tx, &current, rockTokenEventIncident, current.Status, current.Status,
rockTokenSourceIncident, fmt.Sprintf("%s: %s", category, message), false, nil)
return writeAudit(tx, nil, nil, nil, "rock_delivery_incident.open", "rock_token",
fmt.Sprintf("%d", current.ID), map[string]interface{}{
"incident_id": incident.ID,
"order_no": orderNo,
"category": category,
"http_status": httpStatus,
})
})
if err != nil {
return nil, err
}
return &incident, nil
}
func (s *RockService) HasBlockingRockIncident(ctx context.Context, orderNo string) (*model.RockDeliveryIncident, error) {
var incident model.RockDeliveryIncident
err := s.db.WithContext(ctx).
Where("order_no = ? AND status = ? AND blocking_retry = TRUE", strings.TrimSpace(orderNo), model.RockIncidentStatusOpen).
Order("last_seen_at DESC").
First(&incident).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
if err != nil {
return nil, err
}
return &incident, nil
}
func (s *RockService) ListRockTokenIncidents(ctx context.Context, tokenID uint) ([]model.RockDeliveryIncident, error) {
var list []model.RockDeliveryIncident
err := s.db.WithContext(ctx).
Where("token_id = ?", tokenID).
Order("id DESC").
Limit(200).
Find(&list).Error
return list, err
}
func (s *RockService) ListRockDeliveryIncidents(ctx context.Context, status, category string, page, pageSize int) (*RockIncidentListResult, error) {
if page <= 0 {
page = 1
}
if pageSize <= 0 || pageSize > 100 {
pageSize = 20
}
db := s.db.WithContext(ctx).Model(&model.RockDeliveryIncident{})
if v := strings.TrimSpace(status); v != "" {
db = db.Where("status = ?", v)
}
if v := strings.TrimSpace(category); v != "" {
db = db.Where("category = ?", v)
}
var total int64
if err := db.Count(&total).Error; err != nil {
return nil, err
}
var list []model.RockDeliveryIncident
err := db.Order("last_seen_at DESC, id DESC").
Offset((page - 1) * pageSize).
Limit(pageSize).
Find(&list).Error
if err != nil {
return nil, err
}
return &RockIncidentListResult{List: list, Total: total}, nil
}
// ResolveRockDeliveryIncident 只关闭异常事件,不改变 Token 或订单主状态。
func (s *RockService) ResolveRockDeliveryIncident(ctx context.Context, id uint, note string, actorUserID *uint) error {
if id == 0 {
return errors.New("异常记录不存在")
}
now := time.Now()
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var incident model.RockDeliveryIncident
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
Where("id = ? AND status = ?", id, model.RockIncidentStatusOpen).
First(&incident).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return errors.New("异常记录不存在或已处理")
}
return err
}
if err := tx.Model(&incident).Updates(map[string]interface{}{
"status": model.RockIncidentStatusResolved,
"blocking_retry": false,
"resolved_at": now,
"resolved_by": actorUserID,
"resolution_action": "resolve",
"resolution_note": strings.TrimSpace(note),
"updated_at": now,
}).Error; err != nil {
return err
}
return writeAudit(tx, nil, actorUserID, nil, "rock_delivery_incident.resolve", "rock_delivery_incident",
fmt.Sprintf("%d", id), map[string]interface{}{
"token_id": incident.TokenID,
"order_no": incident.OrderNo,
"category": incident.Category,
"note": strings.TrimSpace(note),
})
})
}
// ApproveRockIncidentRetry 只对超时/频控异常临时解除重试阻塞。
// 实际发货结束后必须调用 Complete 或重新记录异常,恢复最终状态。
func (s *RockService) ApproveRockIncidentRetry(
ctx context.Context,
id uint,
note string,
actorUserID *uint,
) (*RockIncidentRetryPlan, error) {
note = strings.TrimSpace(note)
if note == "" {
return nil, errors.New("请填写人工重试说明")
}
now := time.Now()
var plan RockIncidentRetryPlan
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
Where("id = ? AND status = ?", id, model.RockIncidentStatusOpen).
First(&plan.Incident).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return errors.New("异常记录不存在或已处理")
}
return err
}
if !rockIncidentRetryable(plan.Incident.Category) {
return fmt.Errorf("%s 类型异常不允许直接重试,请先核对上游状态", plan.Incident.Category)
}
if err := tx.First(&plan.Token, plan.Incident.TokenID).Error; err != nil {
return errors.New("异常关联的 Token 不存在")
}
if plan.Token.Status != model.RockStatusBound ||
plan.Token.AssignedOrderNo != plan.Incident.OrderNo ||
plan.Token.RedeemedAt != nil ||
rockTokenRedeemPending(&plan.Token) {
return errors.New("Token 当前状态不允许人工重试,请刷新后核对")
}
if strings.TrimSpace(plan.Token.ExpectedUID) == "" {
return errors.New("Token 缺少预期 UID,不能人工重试")
}
if err := tx.Where("order_no = ?", plan.Incident.OrderNo).First(&plan.Order).Error; err != nil {
return errors.New("异常关联的订单不存在")
}
if plan.Order.OrderStatus != model.OrderStatusShipFailed {
return fmt.Errorf("订单当前状态为 %s,不允许人工重试", plan.Order.OrderStatus)
}
var blockingNonRetryable model.RockDeliveryIncident
if err := tx.Where(
"token_id = ? AND order_no = ? AND status = ? AND category NOT IN ?",
plan.Token.ID,
plan.Incident.OrderNo,
model.RockIncidentStatusOpen,
[]string{model.RockIncidentCategoryUpstreamTimeout, model.RockIncidentCategoryRateLimited},
).First(&blockingNonRetryable).Error; err == nil {
return fmt.Errorf("该订单同时存在不可重试异常(%s),请先人工核对", blockingNonRetryable.Category)
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
if err := tx.Model(&model.RockDeliveryIncident{}).
Where("token_id = ? AND order_no = ? AND status = ?",
plan.Token.ID, plan.Incident.OrderNo, model.RockIncidentStatusOpen).
Updates(map[string]interface{}{
"blocking_retry": false,
"updated_at": now,
}).Error; err != nil {
return err
}
return writeAudit(tx, nil, actorUserID, nil, "rock_delivery_incident.approve_retry",
"rock_delivery_incident", fmt.Sprintf("%d", plan.Incident.ID), map[string]interface{}{
"token_id": plan.Token.ID,
"order_no": plan.Incident.OrderNo,
"category": plan.Incident.Category,
"note": note,
})
})
if err != nil {
return nil, err
}
return &plan, nil
}
// CompleteRockIncidentRetry 在上游已受理或已确认发货后,关闭同一订单的开放异常。
func (s *RockService) CompleteRockIncidentRetry(
ctx context.Context,
tokenID uint,
orderNo, note string,
actorUserID *uint,
) error {
now := time.Now()
note = strings.TrimSpace(note)
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
result := tx.Model(&model.RockDeliveryIncident{}).
Where("token_id = ? AND order_no = ? AND status = ?",
tokenID, strings.TrimSpace(orderNo), model.RockIncidentStatusOpen).
Updates(map[string]interface{}{
"status": model.RockIncidentStatusResolved,
"blocking_retry": false,
"resolved_at": now,
"resolved_by": actorUserID,
"resolution_action": "approve_retry",
"resolution_note": note,
"updated_at": now,
})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return errors.New("没有可关闭的开放异常记录")
}
return writeAudit(tx, nil, actorUserID, nil, "rock_delivery_incident.retry_complete",
"rock_token", fmt.Sprintf("%d", tokenID), map[string]interface{}{
"order_no": orderNo,
"note": note,
})
})
}
func resolveOpenRockIncidents(tx *gorm.DB, tokenID uint, action, note string, actorUserID *uint) error {
now := time.Now()
return tx.Model(&model.RockDeliveryIncident{}).
Where("token_id = ? AND status = ?", tokenID, model.RockIncidentStatusOpen).
Updates(map[string]interface{}{
"status": model.RockIncidentStatusResolved,
"blocking_retry": false,
"resolved_at": now,
"resolved_by": actorUserID,
"resolution_action": action,
"resolution_note": strings.TrimSpace(note),
"updated_at": now,
}).Error
}
func rockTokenIncidentFilter(db *gorm.DB, category string) *gorm.DB {
if category == model.RockIncidentCategoryNoEntitlement {
return db.Where(
"EXISTS (SELECT 1 FROM rock_delivery_incidents i WHERE i.token_id = rock_tokens.id AND i.status = 'open' AND i.category = ?)",
model.RockIncidentCategoryNoEntitlement,
)
}
return db.Where(
"EXISTS (SELECT 1 FROM rock_delivery_incidents i WHERE i.token_id = rock_tokens.id AND i.status = 'open' AND i.category <> ?)",
model.RockIncidentCategoryNoEntitlement,
)
}
func (s *RockService) fillRockTokenIncidentFlags(ctx context.Context, tokens []model.RockToken) {
if len(tokens) == 0 {
return
}
ids := make([]uint, 0, len(tokens))
for i := range tokens {
if tokens[i].ID != 0 {
ids = append(ids, tokens[i].ID)
}
}
if len(ids) == 0 {
return
}
type row struct {
TokenID uint
Category string
Message string
LastSeenAt time.Time
IncidentCount int64
}
var rows []row
err := s.db.WithContext(ctx).Raw(`
SELECT DISTINCT ON (token_id)
token_id, category, message, last_seen_at,
COUNT(*) OVER (PARTITION BY token_id) AS incident_count
FROM rock_delivery_incidents
WHERE token_id IN ? AND status = 'open'
ORDER BY token_id, (category = 'no_entitlement') DESC, last_seen_at DESC, id DESC
`, ids).Scan(&rows).Error
if err != nil {
return
}
byID := make(map[uint]row, len(rows))
for _, item := range rows {
byID[item.TokenID] = item
}
for i := range tokens {
item, ok := byID[tokens[i].ID]
if !ok {
continue
}
at := item.LastSeenAt
tokens[i].HasOpenIncident = true
tokens[i].IncidentCategory = item.Category
tokens[i].IncidentMessage = item.Message
tokens[i].IncidentAt = &at
tokens[i].IncidentCount = item.IncidentCount
}
}
@@ -0,0 +1,523 @@
package service
import (
"context"
"errors"
"strings"
"testing"
"time"
"affiliate_dash/internal/model"
)
func TestClassifyRockDeliveryError_TokenMessageWithoutSpaces(t *testing.T) {
category, _, _ := classifyRockDeliveryError(errors.New("上游返回:该Token已兑换"))
if category != model.RockIncidentCategoryTokenRedeemed {
t.Fatalf("应识别无空格 Token 已兑换错误,实际 %s", category)
}
if !strings.Contains(category, "token") {
t.Fatalf("分类内部值应保持稳定,实际 %s", category)
}
}
func TestRockDeliveryIncident_NoEntitlementLifecycle(t *testing.T) {
db := newRockTestDB(t)
svc := NewRockService(db, NewRockClient("https://example.com"))
ctx := context.Background()
token := model.RockToken{
UpstreamTokenID: "incident-1",
Token: "jwt-incident-1",
RedeemCode: "CODE-INCIDENT-1",
Status: model.RockStatusBound,
AssignedOrderNo: "ORDER-INCIDENT-1",
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
cause := &RockUpstreamStatusError{
StatusCode: 400,
Body: `{"error":"用户不存在可以兑换的可用权益"}`,
}
incident, err := svc.RecordRockDeliveryIncident(ctx, &token, token.AssignedOrderNo, cause)
if err != nil {
t.Fatalf("记录异常失败: %v", err)
}
if incident.Category != model.RockIncidentCategoryNoEntitlement || incident.Occurrences != 1 {
t.Fatalf("异常分类或次数错误: %+v", incident)
}
if _, err := svc.RecordRockDeliveryIncident(ctx, &token, token.AssignedOrderNo, cause); err != nil {
t.Fatalf("重复记录异常失败: %v", err)
}
var stored model.RockToken
if err := db.First(&stored, token.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if stored.RedeemLastError == "" {
t.Fatal("Token 应记录最近一次发货错误")
}
blocking, err := svc.HasBlockingRockIncident(ctx, token.AssignedOrderNo)
if err != nil || blocking == nil || blocking.Category != model.RockIncidentCategoryNoEntitlement {
t.Fatalf("应查询到阻塞重试的异常: incident=%+v err=%v", blocking, err)
}
var open model.RockDeliveryIncident
if err := db.Where("token_id = ? AND status = ?", token.ID, model.RockIncidentStatusOpen).
First(&open).Error; err != nil {
t.Fatalf("读取开放异常失败: %v", err)
}
if open.Occurrences != 2 {
t.Fatalf("重复错误应累加次数,实际 %d", open.Occurrences)
}
list, total, err := svc.ListTokens(ctx, 1, 20, "no_entitlement", "")
if err != nil {
t.Fatalf("筛选无权益失败: %v", err)
}
if total != 1 || len(list) != 1 || !list[0].HasOpenIncident {
t.Fatalf("无权益筛选结果错误: total=%d list=%+v", total, list)
}
anomalyList, anomalyTotal, err := svc.ListTokens(ctx, 1, 20, model.RockStatusAnomaly, "")
if err != nil {
t.Fatalf("筛选异常失败: %v", err)
}
if anomalyTotal != 0 || len(anomalyList) != 0 {
t.Fatalf("无权益不能重复归入异常: total=%d list=%+v", anomalyTotal, anomalyList)
}
stats, err := svc.GetStats(ctx)
if err != nil {
t.Fatalf("读取统计失败: %v", err)
}
if stats.NoEntitlementStock != 1 {
t.Fatalf("无权益统计错误: %d", stats.NoEntitlementStock)
}
if stats.AnomalyStock != 0 || stats.UnresolvedAnomalyStock != 0 {
t.Fatalf("无权益不应计入异常统计: anomaly=%d unresolved=%d",
stats.AnomalyStock, stats.UnresolvedAnomalyStock)
}
actor := uint(9)
if err := svc.ResolveRockDeliveryIncident(ctx, open.ID, "已人工核对", &actor); err != nil {
t.Fatalf("处理异常失败: %v", err)
}
blocking, err = svc.HasBlockingRockIncident(ctx, token.AssignedOrderNo)
if err != nil {
t.Fatalf("处理后查询异常失败: %v", err)
}
if blocking != nil {
t.Fatalf("处理后不应继续阻塞重试: %+v", blocking)
}
}
func TestRockDeliveryIncident_OtherErrorsUseAnomalyFilter(t *testing.T) {
db := newRockTestDB(t)
svc := NewRockService(db, NewRockClient("https://example.com"))
ctx := context.Background()
token := model.RockToken{
UpstreamTokenID: "incident-2",
Token: "jwt-incident-2",
RedeemCode: "CODE-INCIDENT-2",
Status: model.RockStatusBound,
AssignedOrderNo: "ORDER-INCIDENT-2",
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
if _, err := svc.RecordRockDeliveryIncident(ctx, &token, token.AssignedOrderNo,
errors.New("上游兑换发货接口返回错误码 400: 同一角色无法重复兑换已兑换过的道具")); err != nil {
t.Fatalf("记录异常失败: %v", err)
}
noEntitlement, totalNoEntitlement, err := svc.ListTokens(ctx, 1, 20, "no_entitlement", "")
if err != nil {
t.Fatalf("筛选无权益失败: %v", err)
}
if totalNoEntitlement != 0 || len(noEntitlement) != 0 {
t.Fatalf("其他错误不应进入无权益: %+v", noEntitlement)
}
anomaly, totalAnomaly, err := svc.ListTokens(ctx, 1, 20, model.RockStatusAnomaly, "")
if err != nil {
t.Fatalf("筛选异常失败: %v", err)
}
if totalAnomaly != 1 || len(anomaly) != 1 || anomaly[0].IncidentCategory != model.RockIncidentCategoryDuplicateItem {
t.Fatalf("其他错误应进入异常: total=%d list=%+v", totalAnomaly, anomaly)
}
}
func TestRockIncidentRetry_RequiresRetryableCategoryAndResolvesAfterApproval(t *testing.T) {
db := newRockTestDB(t)
svc := NewRockService(db, NewRockClient("https://example.com"))
ctx := context.Background()
token := model.RockToken{
UpstreamTokenID: "incident-retry-1",
Token: "jwt-incident-retry-1",
RedeemCode: "CODE-INCIDENT-RETRY-1",
Status: model.RockStatusBound,
AssignedOrderNo: "ORDER-INCIDENT-RETRY-1",
ExpectedUID: "5550001",
ActualRoleID: "5550001",
ActualRoleName: "测试角色",
}
order := model.FulfillmentOrder{
OrderNo: token.AssignedOrderNo,
MerchantID: 1,
OrderStatus: model.OrderStatusShipFailed,
ProductSKU: model.RockSKUBlackTshirt,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
if err := db.Create(&order).Error; err != nil {
t.Fatalf("创建订单失败: %v", err)
}
incident, err := svc.RecordRockIncident(ctx, &token, order.OrderNo,
model.RockIncidentCategoryUpstreamTimeout, "上游超时", 504, "")
if err != nil {
t.Fatalf("记录异常失败: %v", err)
}
actor := uint(9)
plan, err := svc.ApproveRockIncidentRetry(ctx, incident.ID, "已核对上游,允许一次重试", &actor)
if err != nil {
t.Fatalf("批准人工重试失败: %v", err)
}
if plan.Token.ID != token.ID || plan.Order.OrderNo != order.OrderNo {
t.Fatalf("重试计划关联错误: %+v", plan)
}
var unblocked model.RockDeliveryIncident
if err := db.First(&unblocked, incident.ID).Error; err != nil {
t.Fatalf("读取异常失败: %v", err)
}
if unblocked.Status != model.RockIncidentStatusOpen || unblocked.BlockingRetry {
t.Fatalf("批准重试后应暂时解除阻塞: %+v", unblocked)
}
if err := svc.CompleteRockIncidentRetry(ctx, token.ID, order.OrderNo, "重试成功", &actor); err != nil {
t.Fatalf("完成人工重试失败: %v", err)
}
var resolved model.RockDeliveryIncident
if err := db.First(&resolved, incident.ID).Error; err != nil {
t.Fatalf("读取已处理异常失败: %v", err)
}
if resolved.Status != model.RockIncidentStatusResolved ||
resolved.ResolutionAction != "approve_retry" ||
resolved.ResolvedBy == nil || *resolved.ResolvedBy != actor {
t.Fatalf("异常关闭结果错误: %+v", resolved)
}
}
func TestRockIncidentRetry_RejectsNoEntitlement(t *testing.T) {
db := newRockTestDB(t)
svc := NewRockService(db, NewRockClient("https://example.com"))
ctx := context.Background()
token := model.RockToken{
UpstreamTokenID: "incident-retry-2",
Token: "jwt-incident-retry-2",
Status: model.RockStatusBound,
AssignedOrderNo: "ORDER-INCIDENT-RETRY-2",
ExpectedUID: "5550002",
}
order := model.FulfillmentOrder{
OrderNo: token.AssignedOrderNo,
MerchantID: 1,
OrderStatus: model.OrderStatusShipFailed,
ProductSKU: model.RockSKUBlackTshirt,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
if err := db.Create(&order).Error; err != nil {
t.Fatalf("创建订单失败: %v", err)
}
incident, err := svc.RecordRockIncident(ctx, &token, order.OrderNo,
model.RockIncidentCategoryNoEntitlement, "用户不存在可以兑换的可用权益", 400, "")
if err != nil {
t.Fatalf("记录异常失败: %v", err)
}
if _, err := svc.ApproveRockIncidentRetry(ctx, incident.ID, "尝试重试", nil); err == nil {
t.Fatal("无权益异常不允许直接重试")
}
}
func TestRockProviderCancel_UsingTokenReturnsToStock(t *testing.T) {
db := newRockTestDB(t)
svc := NewRockService(db, NewRockClient("https://example.com"))
provider := NewRockProvider(svc)
ctx := context.Background()
usingAt := time.Now().Add(-10 * time.Minute)
token := model.RockToken{
UpstreamTokenID: "cancel-using",
Token: "jwt-cancel-using",
Status: model.RockStatusUsing,
AssignedOrderNo: "ORDER-CANCEL-USING",
ExpectedUID: "5550003",
UsingAt: &usingAt,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
if err := provider.Cancel(ctx, &model.FulfillmentOrder{OrderNo: token.AssignedOrderNo}); err != nil {
t.Fatalf("取消未绑定订单失败: %v", err)
}
var saved model.RockToken
if err := db.First(&saved, token.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if saved.Status != model.RockStatusAvailable || saved.AssignedOrderNo != "" ||
saved.ExpectedUID != "" || saved.UsingAt != nil {
t.Fatalf("未绑定 Token 应安全回库存: %+v", saved)
}
var incidentCount int64
if err := db.Model(&model.RockDeliveryIncident{}).
Where("token_id = ? AND status = ?", token.ID, model.RockIncidentStatusOpen).
Count(&incidentCount).Error; err != nil {
t.Fatalf("统计异常失败: %v", err)
}
if incidentCount != 0 {
t.Fatalf("未绑定订单取消不应生成异常,实际 %d", incidentCount)
}
}
func TestRockProviderCancel_BoundAndRedeemedCreateOrderCancelledIncident(t *testing.T) {
cases := []struct {
name string
status string
}{
{name: "bound", status: model.RockStatusBound},
{name: "redeemed", status: model.RockStatusRedeemed},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
db := newRockTestDB(t)
svc := NewRockService(db, NewRockClient("https://example.com"))
provider := NewRockProvider(svc)
token := model.RockToken{
UpstreamTokenID: "cancel-" + tc.name,
Token: "jwt-cancel-" + tc.name,
Status: tc.status,
AssignedOrderNo: "ORDER-CANCEL-" + tc.name,
ExpectedUID: "5550004",
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
if err := provider.Cancel(context.Background(),
&model.FulfillmentOrder{OrderNo: token.AssignedOrderNo}); err != nil {
t.Fatalf("取消订单失败: %v", err)
}
var stored model.RockToken
if err := db.First(&stored, token.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if stored.Status != tc.status || stored.AssignedOrderNo != token.AssignedOrderNo {
t.Fatalf("已绑定或已发货 Token 不应自动释放: %+v", stored)
}
var incident model.RockDeliveryIncident
if err := db.Where(
"token_id = ? AND order_no = ? AND status = ? AND category = ?",
token.ID,
token.AssignedOrderNo,
model.RockIncidentStatusOpen,
model.RockIncidentCategoryOrderCancelled,
).First(&incident).Error; err != nil {
t.Fatalf("应生成订单取消冲突异常: %v", err)
}
if !incident.BlockingRetry || incident.Message == "" {
t.Fatalf("订单取消冲突异常应阻塞重试并带说明: %+v", incident)
}
})
}
}
func TestReconcileRockOrders_DoesNotBackfillHistoricalIncidents(t *testing.T) {
db := newRockTestDB(t)
svc := NewRockService(db, NewRockClient("https://example.com"))
ctx := context.Background()
tokens := []model.RockToken{
{
UpstreamTokenID: "reconcile-old-failed",
Token: "jwt-reconcile-old-failed",
Status: model.RockStatusBound,
AssignedOrderNo: "ORDER-RECONCILE-OLD-FAILED",
},
{
UpstreamTokenID: "reconcile-old-cancel",
Token: "jwt-reconcile-old-cancel",
Status: model.RockStatusRedeemed,
AssignedOrderNo: "ORDER-RECONCILE-OLD-CANCEL",
},
}
orders := []model.FulfillmentOrder{
{
OrderNo: tokens[0].AssignedOrderNo,
ClientOrderNo: "CLIENT-RECONCILE-OLD-FAILED",
MerchantID: 1,
OrderStatus: model.OrderStatusShipFailed,
FailureReason: "用户不存在可以兑换的可用权益",
ProductSKU: model.RockSKUBlackTshirt,
},
{
OrderNo: tokens[1].AssignedOrderNo,
ClientOrderNo: "CLIENT-RECONCILE-OLD-CANCEL",
MerchantID: 1,
OrderStatus: model.OrderStatusCancelled,
ProductSKU: model.RockSKUBlackTshirt,
},
}
for i := range tokens {
if err := db.Create(&tokens[i]).Error; err != nil {
t.Fatalf("创建历史 Token 失败: %v", err)
}
if err := db.Create(&orders[i]).Error; err != nil {
t.Fatalf("创建历史订单失败: %v", err)
}
}
svc.SetRedeemConfirmHook(func(context.Context, *model.FulfillmentOrder, *model.RockToken) error {
return nil
})
if err := svc.ReconcileRockOrders(ctx); err != nil {
t.Fatalf("执行对账失败: %v", err)
}
var count int64
if err := db.Model(&model.RockDeliveryIncident{}).Count(&count).Error; err != nil {
t.Fatalf("统计异常失败: %v", err)
}
if count != 0 {
t.Fatalf("历史失败/取消记录不应自动回填异常,实际 %d", count)
}
}
func TestPendingOrderDoesNotEnterShipTimeout(t *testing.T) {
db := newRockTestDB(t)
svc := NewFulfillmentService(db, nil)
old := time.Now().Add(-time.Hour)
order := model.FulfillmentOrder{
OrderNo: "ORDER-PENDING-TIMEOUT",
MerchantID: 1,
OrderStatus: model.OrderStatusDelivering,
ResultData: `{"provider_order_stage":"redeem_pending","rock_redeem_pending":true}`,
}
if err := db.Create(&order).Error; err != nil {
t.Fatalf("创建订单失败: %v", err)
}
if err := db.Model(&model.FulfillmentOrder{}).Where("id = ?", order.ID).
UpdateColumn("updated_at", old).Error; err != nil {
t.Fatalf("调整订单时间失败: %v", err)
}
changed, err := svc.MarkProcessingTimeouts(30*time.Minute, 10)
if err != nil {
t.Fatalf("执行超时扫描失败: %v", err)
}
if changed != 0 {
t.Fatalf("pending 订单不应被超时任务标记失败,实际变更 %d", changed)
}
var stored model.FulfillmentOrder
if err := db.First(&stored, order.ID).Error; err != nil {
t.Fatalf("读取订单失败: %v", err)
}
if stored.OrderStatus != model.OrderStatusDelivering {
t.Fatalf("pending 订单应保持 delivering,实际 %s", stored.OrderStatus)
}
}
func TestBlockingIncidentPreventsDeliveryRetry(t *testing.T) {
db := newRockTestDB(t)
svc := NewRockService(db, NewRockClient("https://example.com"))
provider := NewRockProvider(svc)
ctx := context.Background()
token := model.RockToken{
UpstreamTokenID: "incident-block",
Token: "jwt-incident-block",
RedeemCode: "CODE-INCIDENT-BLOCK",
Status: model.RockStatusBound,
AssignedOrderNo: "ORDER-INCIDENT-BLOCK",
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
if _, err := svc.RecordRockDeliveryIncident(ctx, &token, token.AssignedOrderNo,
errors.New("用户不存在可以兑换的可用权益")); err != nil {
t.Fatalf("记录异常失败: %v", err)
}
state, err := provider.PrepareDelivery(ctx, &model.FulfillmentOrder{
OrderNo: token.AssignedOrderNo,
ProductSKU: model.RockSKUBlackTshirt,
}, nil)
if err != nil {
t.Fatalf("准备发货失败: %v", err)
}
if state.CanShip {
t.Fatal("存在开放异常时不得再次调用上游发货")
}
}
func TestReconcileRockOrdersRepairsTokenAndOrderStatus(t *testing.T) {
db := newRockTestDB(t)
svc := NewRockService(db, NewRockClient("https://example.com"))
ctx := context.Background()
redeemedAt := time.Now().Add(-time.Hour)
token := model.RockToken{
UpstreamTokenID: "reconcile-1",
Token: "jwt-reconcile-1",
RedeemCode: "CODE-RECONCILE-1",
Status: model.RockStatusRedeemed,
AssignedOrderNo: "ORDER-RECONCILE-1",
RedeemedAt: &redeemedAt,
}
order := model.FulfillmentOrder{
OrderNo: token.AssignedOrderNo,
ClientOrderNo: "CLIENT-RECONCILE-1",
MerchantID: 1,
OrderStatus: model.OrderStatusPaid,
ProductSKU: model.RockSKUBlackTshirt,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
if err := db.Create(&order).Error; err != nil {
t.Fatalf("创建订单失败: %v", err)
}
svc.SetRedeemConfirmHook(func(ctx context.Context, order *model.FulfillmentOrder, token *model.RockToken) error {
return db.Model(order).Update("order_status", model.OrderStatusDelivered).Error
})
if err := svc.ReconcileRockOrders(ctx); err != nil {
t.Fatalf("订单对账失败: %v", err)
}
var repairedOrder model.FulfillmentOrder
if err := db.First(&repairedOrder, order.ID).Error; err != nil {
t.Fatalf("读取订单失败: %v", err)
}
if repairedOrder.OrderStatus != model.OrderStatusDelivered {
t.Fatalf("redeemed Token 应修复订单为 delivered,实际 %s", repairedOrder.OrderStatus)
}
deliveredAt := time.Now()
legacyToken := model.RockToken{
UpstreamTokenID: "reconcile-2",
Token: "jwt-reconcile-2",
RedeemCode: "CODE-RECONCILE-2",
Status: model.RockStatusBound,
AssignedOrderNo: "ORDER-RECONCILE-2",
RedeemedAt: &redeemedAt,
}
deliveredOrder := model.FulfillmentOrder{
OrderNo: legacyToken.AssignedOrderNo,
ClientOrderNo: "CLIENT-RECONCILE-2",
MerchantID: 1,
OrderStatus: model.OrderStatusDelivered,
DeliveredAt: &deliveredAt,
ProductSKU: model.RockSKUBlackTshirt,
}
if err := db.Create(&legacyToken).Error; err != nil {
t.Fatalf("创建历史 Token 失败: %v", err)
}
if err := db.Create(&deliveredOrder).Error; err != nil {
t.Fatalf("创建已交付订单失败: %v", err)
}
if err := svc.ReconcileRockOrders(ctx); err != nil {
t.Fatalf("Token 对账失败: %v", err)
}
var repairedToken model.RockToken
if err := db.First(&repairedToken, legacyToken.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if repairedToken.Status != model.RockStatusRedeemed {
t.Fatalf("已交付订单的 bound Token 应修复为 redeemed,实际 %s", repairedToken.Status)
}
}
+30 -5
View File
@@ -10,6 +10,7 @@ import (
"time"
"affiliate_dash/internal/model"
"affiliate_dash/internal/pkg/timeutil"
)
func TestUpdateConfig_BoundWarnDaysAndAutoRefresh(t *testing.T) {
@@ -65,7 +66,15 @@ func TestListTokens_OverdueFilters(t *testing.T) {
boundOld := model.RockToken{UpstreamTokenID: "overdue-bound", RedeemCode: "C3", Status: model.RockStatusBound, BoundAt: &old}
boundFresh := model.RockToken{UpstreamTokenID: "fresh-bound", RedeemCode: "C4", Status: model.RockStatusBound, BoundAt: &fresh}
redeemedOld := model.RockToken{UpstreamTokenID: "old-redeemed", RedeemCode: "C5", Status: model.RockStatusRedeemed, BoundAt: &old}
for _, token := range []*model.RockToken{&usingOld, &usingFresh, &boundOld, &boundFresh, &redeemedOld} {
requestedAt := now.Add(-time.Minute)
boundPending := model.RockToken{
UpstreamTokenID: "overdue-bound-pending",
RedeemCode: "C6",
Status: model.RockStatusBound,
BoundAt: &old,
RedeemRequestedAt: &requestedAt,
}
for _, token := range []*model.RockToken{&usingOld, &usingFresh, &boundOld, &boundFresh, &redeemedOld, &boundPending} {
if err := db.Create(token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
@@ -96,6 +105,15 @@ func TestListTokens_OverdueFilters(t *testing.T) {
t.Fatalf("过期预警统计不符: using_overdue=%d using_danger=%d bound_overdue=%d",
stats.UsingOverdueStock, stats.UsingDangerStock, stats.BoundOverdueStock)
}
all, _, err := svc.ListTokens(ctx, 1, 20, "", "")
if err != nil {
t.Fatalf("查询全部 Token 失败: %v", err)
}
for _, token := range all {
if token.ID == boundPending.ID && token.BoundOverdue {
t.Fatal("发货确认中的 Token 不应进入绑定后未发货")
}
}
}
// TestListTokens_OverdueBoundaryInclusive 锁定「满 2 天 / 满 3 天」的含边界口径:
@@ -211,15 +229,17 @@ func TestRefreshStaleTokens_SkipsWhenAutoRefreshDisabled(t *testing.T) {
}
}
func TestRefreshAllStaleTokens_FixesMismatch(t *testing.T) {
func TestRefreshAllStaleTokens_KeepsUnresolvedMismatch(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")
bindingAt := time.Now().In(timeutil.Location()).Add(-30 * time.Minute).Format("2006-01-02 15:04:05")
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"token": map[string]interface{}{
"status": "bound",
"bound_account": "痛写心扉",
"bound_account_id": "3132710334",
"bind_changed_at": bindingAt,
},
})
}))
@@ -228,11 +248,13 @@ func TestRefreshAllStaleTokens_FixesMismatch(t *testing.T) {
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
usingAt := time.Now().Add(-time.Hour)
token := model.RockToken{
UpstreamTokenID: "scan-on",
RedeemCode: "CODE-SCAN-ON",
Status: model.RockStatusMismatch,
ExpectedUID: "1111111111",
UsingAt: &usingAt,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
@@ -246,7 +268,7 @@ func TestRefreshAllStaleTokens_FixesMismatch(t *testing.T) {
if err != nil {
t.Fatalf("一键刷新失败: %v", err)
}
if result.Scanned != 1 || result.Fixed != 1 || result.Remaining != 0 || result.Failed != 0 {
if result.Scanned != 1 || result.Fixed != 0 || result.Remaining != 1 || result.Failed != 0 {
t.Fatalf("一键刷新结果不符: %+v", result)
}
@@ -254,8 +276,8 @@ func TestRefreshAllStaleTokens_FixesMismatch(t *testing.T) {
if err := db.First(&saved, token.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if saved.Status != model.RockStatusBound || saved.MismatchAt != nil {
t.Fatalf("应修复为 bound 并清空 mismatch 标记,实际 %+v", saved)
if saved.Status != model.RockStatusMismatch || saved.MismatchAt == nil {
t.Fatalf("UID 不一致时应保持 mismatch,实际 %+v", saved)
}
}
@@ -271,6 +293,7 @@ func TestRefreshAllStaleTokens_IncludesOverdueTokens(t *testing.T) {
"status": "bound",
"bound_account": "新角色",
"bound_account_id": "10001",
"bind_changed_at": time.Now().AddDate(0, 0, -10).Add(time.Hour).In(timeutil.Location()).Format("2006-01-02 15:04:05"),
},
})
case "/api/redeem/CODE-BOUND-OLD/refresh":
@@ -302,6 +325,7 @@ func TestRefreshAllStaleTokens_IncludesOverdueTokens(t *testing.T) {
UpstreamTokenID: "stale-bound",
RedeemCode: "CODE-BOUND-OLD",
Status: model.RockStatusBound,
UsingAt: &old,
BoundAt: &old,
}
for _, token := range []*model.RockToken{&usingToken, &boundToken} {
@@ -347,6 +371,7 @@ func TestRefreshStaleTokens_IncludesOverdueWhenEnabled(t *testing.T) {
"status": "bound",
"bound_account": "自动刷新角色",
"bound_account_id": "20001",
"bind_changed_at": time.Now().AddDate(0, 0, -10).Add(time.Hour).In(timeutil.Location()).Format("2006-01-02 15:04:05"),
},
})
}))
+11 -14
View File
@@ -30,7 +30,7 @@ func (s *RockService) importPackageDetail(ctx context.Context, detail *RockPacka
continue
}
upstreamID := strconv.FormatInt(pt.ID, 10)
status := normalizeRockPackageStatus(pt.Status)
status := rockStatusFromPackageToken(pt)
raw, marshalErr := json.Marshal(pt)
if marshalErr != nil {
raw = []byte(`{}`)
@@ -174,22 +174,19 @@ func (s *RockService) importPackageDetail(ctx context.Context, detail *RockPacka
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:
// rockStatusFromPackageToken 以真实事件字段推导状态,避免 status=bound 的历史残留污染库存。
func rockStatusFromPackageToken(token RockPackageToken) string {
if strings.TrimSpace(token.RedeemedAt) != "" {
return model.RockStatusRedeemed
case model.RockStatusMismatch:
return model.RockStatusMismatch
default:
// 未知上游状态默认落 abnormal,避免被错误分配为可用库存。
}
if strings.TrimSpace(token.BindChangedAt) != "" {
return model.RockStatusBound
}
if strings.TrimSpace(token.BindChangeType) != "" {
// 只有事件类型没有时间,无法判定真实状态,保守落异常等待人工核对。
return model.RockStatusAbnormal
}
return model.RockStatusAvailable
}
func parseRockPackageTime(raw string) *time.Time {
+13 -7
View File
@@ -135,6 +135,9 @@ func (s *RockService) ResetMismatchToken(ctx context.Context, tokenID uint, acto
}
writeRockTokenStatusChange(tx, &token, oldStatus, model.RockStatusAvailable,
rockTokenSourceReset, "人工解冻重置回库存", false, actorUserID)
if err := resolveOpenRockIncidents(tx, token.ID, "reset", "人工解冻重置回库存", actorUserID); err != nil {
return err
}
return writeAudit(tx, nil, actorUserID, nil, "rock_token.reset", "rock_token",
fmt.Sprintf("%d", token.ID), map[string]interface{}{
"from_status": oldStatus,
@@ -164,15 +167,12 @@ func (s *RockService) UpdateTokenRemark(ctx context.Context, tokenID uint, remar
return nil, fmt.Errorf("发货确认中的 Token 只能标记为已上报失效,不能改回其他状态")
}
switch trimmedStatus {
case model.RockStatusAbnormal, model.RockStatusAvailable, model.RockStatusMismatch:
case model.RockStatusAbnormal:
updates["status"] = trimmedStatus
token.Status = trimmedStatus
if trimmedStatus != oldStatus {
updates["restocked_at"] = nil // 人工改状态不属于「释放回库存」,不产生优先权
}
if trimmedStatus == model.RockStatusAvailable {
updates["using_at"] = nil
}
if trimmedStatus == model.RockStatusAbnormal {
s.refreshCache.Delete(token.UpstreamTokenID)
}
@@ -187,12 +187,11 @@ func (s *RockService) UpdateTokenRemark(ctx context.Context, tokenID uint, remar
}
writeRockTokenStatusChange(tx, &token, oldStatus, token.Status,
rockTokenSourceManual, "人工更新备注: "+trimmedRemark, false, actorUserID)
_ = writeAudit(tx, nil, actorUserID, nil, "rock_token.update_remark", "rock_token", fmt.Sprintf("%d", token.ID), map[string]interface{}{
return writeAudit(tx, nil, actorUserID, nil, "rock_token.update_remark", "rock_token", fmt.Sprintf("%d", token.ID), map[string]interface{}{
"remark": token.Remark,
"from_status": oldStatus,
"to_status": token.Status,
})
return nil
})
if err != nil {
return nil, err
@@ -207,5 +206,12 @@ func (s *RockService) MarkTokenAbnormal(ctx context.Context, tokenID uint, reaso
if remark == "" {
remark = "已上报上游 (失效作废)"
}
return s.UpdateTokenRemark(ctx, tokenID, remark, model.RockStatusAbnormal, actorUserID)
token, err := s.UpdateTokenRemark(ctx, tokenID, remark, model.RockStatusAbnormal, actorUserID)
if err != nil {
return nil, err
}
if err := resolveOpenRockIncidents(s.db.WithContext(ctx), tokenID, "mark_abnormal", remark, actorUserID); err != nil {
return nil, err
}
return token, nil
}
+103 -53
View File
@@ -126,21 +126,27 @@ func (s *RockService) RedeemTokenForOrder(ctx context.Context, orderNo, sku, exp
if isRockRateLimitError(err) {
markRockUpstreamRateLimited()
}
return nil, fmt.Errorf("调用上游兑换失败: %w", err)
wrapped := fmt.Errorf("调用上游兑换失败: %w", err)
if _, incidentErr := s.RecordRockDeliveryIncident(ctx, &token, orderNo, wrapped); incidentErr != nil {
return nil, fmt.Errorf("%w;记录发货异常失败: %v", wrapped, incidentErr)
}
return nil, wrapped
}
// 上游已受理但官方仍在异步处理:进入 redeeming,等待后台轮询确认。
if redeemResp.IsPending() {
msg := redeemResp.FailureMessage("官方处理中,请稍后刷新查看结果")
now := time.Now()
token = s.markTokenRedeemPending(ctx, token, targetSkin, msg, recorder)
var pendingErr error
token, pendingErr = s.markTokenRedeemPending(ctx, token, targetSkin, msg, recorder)
if pendingErr != nil {
return nil, fmt.Errorf("上游已受理兑换,但记录发货确认状态失败: %w", pendingErr)
}
if recorder != nil {
recorder.Internal(ctx, "redeem_result", "pending", msg, map[string]interface{}{
"upstream_status": redeemResp.Status,
"pending": true,
})
}
_ = now
return &RockRedeemResult{Token: &token, Response: redeemResp, Pending: true, Message: msg, TraceID: recorderTraceID(recorder)}, nil
}
@@ -155,10 +161,14 @@ func (s *RockService) RedeemTokenForOrder(ctx context.Context, orderNo, sku, exp
}
// 刷新失败不直接否认可信的成功响应,但也不立即写终态;交给后台轮询确认。
msg := "上游已受理,正在等待发货结果确认"
token = s.markTokenRedeemPending(ctx, token, targetSkin, msg, recorder)
token, pendingErr := s.markTokenRedeemPending(ctx, token, targetSkin, msg, recorder)
if pendingErr != nil {
return nil, fmt.Errorf("上游已受理兑换,但记录发货确认状态失败: %w", pendingErr)
}
return &RockRedeemResult{Token: &token, Response: redeemResp, Pending: true, Message: msg, TraceID: recorderTraceID(recorder)}, nil
}
if ref != nil && !rockUpstreamRedeemed(ref) {
deliveryEvidence := rockCurrentCycleDelivery(&token, ref)
if ref != nil && !deliveryEvidence.Current && !deliveryEvidence.Ambiguous {
msg := "上游返回成功但尚未确认发货,正在等待官方结果"
if recorder != nil {
recorder.Internal(ctx, "redeem_confirm", "pending", msg, map[string]interface{}{
@@ -166,22 +176,23 @@ func (s *RockService) RedeemTokenForOrder(ctx context.Context, orderNo, sku, exp
"redeemed_at": ref.Token.RedeemedAt,
})
}
token = s.markTokenRedeemPending(ctx, token, targetSkin, msg, recorder)
token, pendingErr := s.markTokenRedeemPending(ctx, token, targetSkin, msg, recorder)
if pendingErr != nil {
return nil, fmt.Errorf("上游已受理兑换,但记录发货确认状态失败: %w", pendingErr)
}
return &RockRedeemResult{Token: &token, Response: redeemResp, Pending: true, Message: msg, TraceID: recorderTraceID(recorder)}, nil
}
fulfillID := extractRockFulfillID(ref, rockClaimBaseline(&token))
fulfillID := deliveryEvidence.FulfillID
redeemedAt := time.Now()
if ref != nil {
if parsed, ok := parseRockTime(ref.Token.RedeemedAt); ok {
redeemedAt = parsed
}
if !deliveryEvidence.At.IsZero() {
redeemedAt = deliveryEvidence.At
}
// 状态跃迁至 redeemed(只有拿到明确成功/明显履约结果才执行)
fromStatus := token.Status
now := time.Now()
_ = s.db.WithContext(ctx).Unscoped().Model(&token).Updates(map[string]interface{}{
if err := s.db.WithContext(ctx).Unscoped().Model(&token).Updates(map[string]interface{}{
"status": model.RockStatusRedeemed,
"upstream_fulfill_id": fulfillID,
"target_skin": targetSkin,
@@ -194,7 +205,9 @@ func (s *RockService) RedeemTokenForOrder(ctx context.Context, orderNo, sku, exp
"redeem_last_message": "",
"redeem_last_error": "",
"redeem_timeout_warned_at": nil,
}).Error
}).Error; err != nil {
return nil, fmt.Errorf("上游已发货,但更新本地 Token 状态失败: %w", err)
}
writeRockTokenStatusChange(s.db.WithContext(ctx), &token, fromStatus, model.RockStatusRedeemed,
rockTokenSourceRedeem, "自动发货: "+targetSkin, false, nil)
if recorder != nil {
@@ -216,14 +229,14 @@ func (s *RockService) RedeemTokenForOrder(ctx context.Context, orderNo, sku, exp
}
// markTokenRedeemPending 记录「发货确认中」标记(不改变 Token 状态)。
func (s *RockService) markTokenRedeemPending(ctx context.Context, token model.RockToken, targetSkin, message string, recorder *RockDeliveryRecorder) model.RockToken {
func (s *RockService) markTokenRedeemPending(ctx context.Context, token model.RockToken, targetSkin, message string, recorder *RockDeliveryRecorder) (model.RockToken, error) {
now := time.Now()
requestedAt := token.RedeemRequestedAt
if requestedAt == nil {
requestedAt = &now
}
nextRetry := now.Add(30 * time.Second)
_ = s.db.WithContext(ctx).Unscoped().Model(&token).Updates(map[string]interface{}{
if err := s.db.WithContext(ctx).Unscoped().Model(&token).Updates(map[string]interface{}{
"target_skin": targetSkin,
"target_currency": "幸运币",
"redeem_requested_at": requestedAt,
@@ -233,7 +246,9 @@ func (s *RockService) markTokenRedeemPending(ctx context.Context, token model.Ro
"redeem_last_message": strings.TrimSpace(message),
"redeem_last_error": "",
"updated_at": now,
}).Error
}).Error; err != nil {
return token, err
}
if recorder != nil {
recorder.Internal(ctx, "redeem_pending", "pending", strings.TrimSpace(message), map[string]interface{}{
"status": token.Status,
@@ -249,7 +264,7 @@ func (s *RockService) markTokenRedeemPending(ctx context.Context, token model.Ro
token.RedeemNextRetryAt = &nextRetry
token.RedeemLastMessage = strings.TrimSpace(message)
token.RedeemLastError = ""
return token
return token, nil
}
// rockTokenRedeemPending 判断 Token 是否处于「发货确认中」:
@@ -271,19 +286,8 @@ func recorderTraceID(recorder *RockDeliveryRecorder) string {
return recorder.TraceID
}
// rockUpstreamRedeemed 判断上游是否已明确完成发货。
func rockUpstreamRedeemed(ref *RockRefreshResponse) bool {
if ref == nil {
return false
}
if strings.EqualFold(strings.TrimSpace(ref.Token.Status), model.RockStatusRedeemed) {
return true
}
return strings.TrimSpace(ref.Token.RedeemedAt) != ""
}
// extractRockFulfillID 从上游客服端返回的 deliveryList 中取本次履约流水。
// 优先取本次领取之后产生的记录;都为历史记录时退回最大流水号。
// 只取本次领取之后产生的记录;历史流水不能作为本轮发货依据。
func extractRockFulfillID(ref *RockRefreshResponse, baseline time.Time) int64 {
if ref == nil || len(ref.DeliveryList) == 0 {
return 0
@@ -306,6 +310,9 @@ func extractRockFulfillID(ref *RockRefreshResponse, baseline time.Time) int64 {
if best > 0 {
return best
}
if !baseline.IsZero() {
return 0
}
for _, item := range ref.DeliveryList {
if item.FulfillID > best {
best = item.FulfillID
@@ -347,8 +354,14 @@ func (s *RockService) checkReleaseToken(ctx context.Context, tokenID uint) (*mod
if err := s.db.WithContext(ctx).First(&token, tokenID).Error; err != nil {
return nil, nil, err
}
if token.Status != model.RockStatusUsing {
return nil, nil, fmt.Errorf("仅扫码使用中的链接可以释放回库存")
if token.Status != model.RockStatusUsing && token.Status != model.RockStatusBound {
return nil, nil, fmt.Errorf("仅使用中或已绑定的链接可以释放回库存")
}
if rockTokenRedeemPending(&token) {
return nil, nil, fmt.Errorf("发货确认中的 Token 禁止释放回库存")
}
if token.RedeemedAt != nil {
return nil, nil, fmt.Errorf("本地已有发货记录的 Token 禁止释放回库存")
}
cfg, err := s.GetConfig(ctx)
@@ -385,6 +398,10 @@ func (s *RockService) checkReleaseToken(ctx context.Context, tokenID uint) (*mod
upstreamStatus := strings.ToLower(strings.TrimSpace(ref.Token.Status))
roleName := ref.GetRoleName()
roleID := ref.GetRoleID()
deliveryEvidence := rockCurrentCycleDelivery(&token, ref)
currentBound, _ := rockCurrentCycleBinding(&token, ref)
baselineUnknown := rockClaimBaseline(&token).IsZero()
matchesInitialSnapshot := rockBindingMatchesInitialSnapshot(&token, ref)
switch {
case upstreamStatus == model.RockStatusAbnormal:
return &token, &RockReleaseCheckResult{
@@ -393,20 +410,32 @@ func (s *RockService) checkReleaseToken(ctx context.Context, tokenID uint) (*mod
RoleID: roleID,
Message: "上游状态异常(已上报失效)",
}, nil
case upstreamStatus == model.RockStatusRedeemed || strings.TrimSpace(ref.Token.RedeemedAt) != "":
case deliveryEvidence.Current || deliveryEvidence.Ambiguous:
message := "上游已发货"
if deliveryEvidence.Ambiguous {
message = "上游显示已发货,但无法确认是否属于本轮,禁止释放"
}
return &token, &RockReleaseCheckResult{
UpstreamStatus: "redeemed",
RoleName: roleName,
RoleID: roleID,
Message: "上游已发货",
Message: message,
}, nil
case rockBuyerNewlyBound(&token, ref):
case currentBound:
return &token, &RockReleaseCheckResult{
UpstreamStatus: "bound",
RoleName: roleName,
RoleID: roleID,
Message: fmt.Sprintf("上游已扫码绑定角色「%s」(%s)", displayRockRole(roleName, roleID), roleID),
}, nil
case baselineUnknown && !matchesInitialSnapshot &&
(ref.HasBindingSignal() || strings.TrimSpace(rockBindingChangedAt(ref)) != ""):
return &token, &RockReleaseCheckResult{
UpstreamStatus: "bound",
RoleName: roleName,
RoleID: roleID,
Message: "上游存在绑定信息,但缺少本轮 using_at,无法自动判断,请人工核对后强制释放",
}, nil
default:
return &token, &RockReleaseCheckResult{
CanRelease: true,
@@ -427,6 +456,10 @@ func (s *RockService) ReleaseUnboundUsingToken(ctx context.Context, tokenID uint
if !check.CanRelease && !force {
return &RockReleaseRequiresForceError{Check: *check}
}
switch check.UpstreamStatus {
case "redeemed", "abnormal", "not_found", "check_failed":
return fmt.Errorf("上游状态为 %s,禁止释放回库存", check.UpstreamStatus)
}
reason := "人工释放(上游未绑定)"
if force {
@@ -440,33 +473,50 @@ func (s *RockService) ReleaseUnboundUsingToken(ctx context.Context, tokenID uint
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&token, tokenID).Error; err != nil {
return err
}
if token.Status != model.RockStatusUsing {
return fmt.Errorf("仅扫码使用中的链接可以释放回库存")
if token.Status != model.RockStatusUsing && token.Status != model.RockStatusBound {
return fmt.Errorf("仅使用中或已绑定的链接可以释放回库存")
}
if rockTokenRedeemPending(&token) {
return fmt.Errorf("发货确认中的 Token 禁止释放回库存")
}
if token.RedeemedAt != nil {
return fmt.Errorf("本地已有发货记录的 Token 禁止释放回库存")
}
fromStatus := token.Status
orderNo = token.AssignedOrderNo
now := time.Now()
if err := tx.Model(&token).Updates(map[string]interface{}{
"status": model.RockStatusAvailable,
"assigned_order_no": "",
"expected_uid": "",
"actual_role_name": "",
"actual_role_id": "",
"actual_channel": "",
"initial_bound_snapshot": "",
"mismatch_at": nil,
"mismatch_reason": "",
"bound_at": nil,
"using_at": nil,
"redeemed_at": nil,
"upstream_fulfill_id": 0,
"restocked_at": now, // 人工释放回库存:下次分配优先使用
"updated_at": now,
"status": model.RockStatusAvailable,
"assigned_order_no": "",
"expected_uid": "",
"actual_role_name": "",
"actual_role_id": "",
"actual_channel": "",
"initial_bound_snapshot": "",
"mismatch_at": nil,
"mismatch_reason": "",
"bound_at": nil,
"using_at": nil,
"redeemed_at": nil,
"upstream_fulfill_id": 0,
"restocked_at": now, // 人工释放回库存:下次分配优先使用
"redeem_requested_at": nil,
"redeem_checked_at": nil,
"redeem_attempts": 0,
"redeem_next_retry_at": nil,
"redeem_last_message": "",
"redeem_last_error": "",
"redeem_timeout_warned_at": nil,
"updated_at": now,
}).Error; err != nil {
return err
}
released = token
writeRockTokenStatusChange(tx, &token, model.RockStatusUsing, model.RockStatusAvailable,
writeRockTokenStatusChange(tx, &token, fromStatus, model.RockStatusAvailable,
rockTokenSourceRelease, reason, force, actorUserID)
if err := resolveOpenRockIncidents(tx, tokenID, "release", reason, actorUserID); err != nil {
return err
}
return writeAudit(tx, nil, actorUserID, nil, "rock_token.release", "rock_token",
fmt.Sprintf("%d", token.ID), map[string]interface{}{
"order_no": orderNo,
+142 -21
View File
@@ -43,6 +43,7 @@ func (s *RockService) SetRedeemConfirmHook(hook func(ctx context.Context, order
// 避免把 0666.pro 的公开刷新接口打成持续频控。
func (s *RockService) RunRedeemConfirmWorker(ctx context.Context) {
lastCleanup := time.Now()
lastReconcile := time.Time{}
ticker := time.NewTicker(rockRedeemPollInterval)
defer ticker.Stop()
for {
@@ -56,6 +57,12 @@ func (s *RockService) RunRedeemConfirmWorker(ctx context.Context) {
if _, err := s.ProcessRedeemingTokens(ctx); err != nil && !errors.Is(err, errRockRedeemBusy) {
log.Printf("[Rock] 发货结果确认任务失败: %v", err)
}
if time.Since(lastReconcile) >= 2*time.Minute {
if err := s.ReconcileRockOrders(ctx); err != nil {
log.Printf("[Rock] 订单对账失败: %v", err)
}
lastReconcile = time.Now()
}
if time.Since(lastCleanup) >= 24*time.Hour {
s.CleanupRockDeliveryLogs(ctx)
lastCleanup = time.Now()
@@ -64,6 +71,93 @@ func (s *RockService) RunRedeemConfirmWorker(ctx context.Context) {
}
}
// ReconcileRockOrders 修复 Token 与订单状态之间的已知不一致:
// 1. Token 已 redeemed,但订单仍 paid/delivering/ship_failed,重新同步订单;
// 2. 订单已 delivered,Token 仍 bound 且已有 redeemed_at,修复 Token 终态;
//
// 历史失败订单和取消冲突不在这里回填;新异常由实际发货/取消流程直接记录。
func (s *RockService) ReconcileRockOrders(ctx context.Context) error {
if s.redeemConfirmHook == nil {
return nil
}
type row struct {
TokenID uint
}
var redeemRows []row
if err := s.db.WithContext(ctx).Raw(`
SELECT t.id AS token_id
FROM rock_tokens t
JOIN fulfillment_orders o ON o.order_no = t.assigned_order_no
WHERE t.deleted_at IS NULL
AND t.status = 'redeemed'
AND o.deleted_at IS NULL
AND o.order_status IN ('paid', 'delivering', 'ship_failed')
ORDER BY t.redeemed_at ASC, t.id ASC
LIMIT 50
`).Scan(&redeemRows).Error; err != nil {
return err
}
for _, item := range redeemRows {
if err := s.reconcileRedeemedOrder(ctx, item.TokenID); err != nil {
log.Printf("[Rock] 对账 redeemed Token %d 失败: %v", item.TokenID, err)
}
}
// 修复历史“订单已交付但 Token 仍 bound”的记录。
var boundRows []row
if err := s.db.WithContext(ctx).Raw(`
SELECT t.id AS token_id
FROM rock_tokens t
JOIN fulfillment_orders o ON o.order_no = t.assigned_order_no
WHERE t.deleted_at IS NULL
AND t.status = 'bound'
AND t.redeemed_at IS NOT NULL
AND o.deleted_at IS NULL
AND o.order_status = 'delivered'
ORDER BY t.id ASC
LIMIT 50
`).Scan(&boundRows).Error; err != nil {
return err
}
for _, item := range boundRows {
var token model.RockToken
if err := s.db.WithContext(ctx).First(&token, item.TokenID).Error; err != nil {
continue
}
oldStatus := token.Status
if err := s.db.WithContext(ctx).Model(&token).Updates(map[string]interface{}{
"status": model.RockStatusRedeemed,
"updated_at": time.Now(),
}).Error; err != nil {
continue
}
writeRockTokenStatusChange(s.db.WithContext(ctx), &token, oldStatus, model.RockStatusRedeemed,
rockTokenSourceRedeem, "订单对账修复:订单已交付", false, nil)
}
return nil
}
func (s *RockService) reconcileRedeemedOrder(ctx context.Context, tokenID uint) error {
var token model.RockToken
if err := s.db.WithContext(ctx).First(&token, tokenID).Error; err != nil {
return err
}
var order model.FulfillmentOrder
if err := s.db.WithContext(ctx).Where("order_no = ?", token.AssignedOrderNo).First(&order).Error; err != nil {
return err
}
if s.redeemConfirmHook == nil {
return nil
}
if err := s.redeemConfirmHook(ctx, &order, &token); err != nil {
_, _ = s.RecordRockIncident(ctx, &token, token.AssignedOrderNo,
model.RockIncidentCategoryOther,
"Token 已发货但订单同步失败: "+err.Error(), 0, "")
return err
}
return nil
}
var errRockRedeemBusy = errors.New("发货结果确认任务正在执行")
// ProcessRedeemingTokens 扫描并确认一批「发货确认中」的 Token。
@@ -138,14 +232,18 @@ func (s *RockService) confirmRedeemingToken(ctx context.Context, token *model.Ro
if isRockRateLimitError(err) {
markRockUpstreamRateLimited()
}
s.bumpRedeemRetry(ctx, token, err.Error(), false)
if retryErr := s.bumpRedeemRetry(ctx, token, err.Error(), false); retryErr != nil {
return false, false, fmt.Errorf("刷新失败且记录重试状态失败: %w", retryErr)
}
if recorder != nil {
recorder.Internal(ctx, "redeem_confirm", "failed", err.Error(), nil)
}
return false, false, err
}
if ref == nil {
s.bumpRedeemRetry(ctx, token, "上游未返回刷新结果", false)
if retryErr := s.bumpRedeemRetry(ctx, token, "上游未返回刷新结果", false); retryErr != nil {
return false, false, retryErr
}
return false, false, errors.New("上游未返回刷新结果")
}
@@ -154,18 +252,24 @@ func (s *RockService) confirmRedeemingToken(ctx context.Context, token *model.Ro
// 上游明确失效:终止处理并标记异常。
if upstreamStatus == model.RockStatusAbnormal {
s.markRedeemingAbnormal(ctx, token, "上游状态异常(发货确认)", recorder)
if err := s.markRedeemingAbnormal(ctx, token, "上游状态异常(发货确认)", recorder); err != nil {
return false, false, err
}
return false, false, nil
}
if rockUpstreamRedeemed(ref) {
fulfillID := extractRockFulfillID(ref, rockClaimBaseline(token))
redeemedAt := now
if parsed, ok := parseRockTime(ref.Token.RedeemedAt); ok {
redeemedAt = parsed
deliveryEvidence := rockCurrentCycleDelivery(token, ref)
if deliveryEvidence.Current || deliveryEvidence.Ambiguous {
fulfillID := deliveryEvidence.FulfillID
if fulfillID == 0 {
fulfillID = extractRockFulfillID(ref, rockClaimBaseline(token))
}
redeemedAt := deliveryEvidence.At
if redeemedAt.IsZero() {
redeemedAt = now
}
fromStatus := token.Status
_ = s.db.WithContext(ctx).Unscoped().Model(token).Updates(map[string]interface{}{
if err := s.db.WithContext(ctx).Unscoped().Model(token).Updates(map[string]interface{}{
"status": model.RockStatusRedeemed,
"upstream_fulfill_id": fulfillID,
"redeemed_at": redeemedAt,
@@ -175,17 +279,24 @@ func (s *RockService) confirmRedeemingToken(ctx context.Context, token *model.Ro
"redeem_last_error": "",
"redeem_timeout_warned_at": nil,
"updated_at": now,
}).Error
}).Error; err != nil {
return false, false, fmt.Errorf("上游确认发货,但更新 Token 终态失败: %w", err)
}
message := "上游确认发货"
if deliveryEvidence.Ambiguous {
message = "上游确认发货,但兑换时间无法核对,已按已发货处理"
}
writeRockTokenStatusChange(s.db.WithContext(ctx), token, fromStatus, model.RockStatusRedeemed,
rockTokenSourceRedeem, "上游确认发货", false, nil)
rockTokenSourceRedeem, message, false, nil)
token.Status = model.RockStatusRedeemed
token.UpstreamFulfillID = fulfillID
token.RedeemedAt = &redeemedAt
if recorder != nil {
recorder.Internal(ctx, "state_change", "success", "上游确认发货", map[string]interface{}{
recorder.Internal(ctx, "state_change", "success", message, map[string]interface{}{
"upstream_status": ref.Token.Status,
"upstream_redeemed_at": ref.Token.RedeemedAt,
"upstream_fulfill_id": fulfillID,
"ambiguous": deliveryEvidence.Ambiguous,
})
}
s.syncDeliveredOrder(ctx, token, recorder)
@@ -197,17 +308,21 @@ func (s *RockService) confirmRedeemingToken(ctx context.Context, token *model.Ro
if waitingMessage == "" {
waitingMessage = "官方处理中,请稍后刷新查看结果"
}
_ = s.db.WithContext(ctx).Unscoped().Model(token).Updates(map[string]interface{}{
if err := s.db.WithContext(ctx).Unscoped().Model(token).Updates(map[string]interface{}{
"redeem_checked_at": now,
"redeem_last_message": waitingMessage,
"redeem_last_error": "",
"updated_at": now,
}).Error
}).Error; err != nil {
return false, false, fmt.Errorf("更新发货确认状态失败: %w", err)
}
token.RedeemCheckedAt = &now
token.RedeemLastMessage = waitingMessage
attempts := token.RedeemAttempts
s.bumpRedeemRetry(ctx, token, "", true)
if err := s.bumpRedeemRetry(ctx, token, "", true); err != nil {
return false, false, err
}
if recorder != nil {
recorder.Internal(ctx, "redeem_confirm", "pending", waitingMessage, map[string]interface{}{
"upstream_status": ref.Token.Status,
@@ -259,9 +374,9 @@ func (s *RockService) refreshRedeemStatus(ctx context.Context, token *model.Rock
return nil, errors.New("缺少兑换码与上游 Token ID,无法确认发货结果")
}
func (s *RockService) bumpRedeemRetry(ctx context.Context, token *model.RockToken, lastError string, countAttempt bool) {
func (s *RockService) bumpRedeemRetry(ctx context.Context, token *model.RockToken, lastError string, countAttempt bool) error {
if token == nil {
return
return errors.New("Token 不存在")
}
attempts := token.RedeemAttempts
if countAttempt {
@@ -290,21 +405,26 @@ func (s *RockService) bumpRedeemRetry(ctx context.Context, token *model.RockToke
if strings.TrimSpace(lastError) != "" {
updates["redeem_last_error"] = lastError
}
_ = s.db.WithContext(ctx).Unscoped().Model(token).Updates(updates).Error
if err := s.db.WithContext(ctx).Unscoped().Model(token).Updates(updates).Error; err != nil {
return err
}
token.RedeemAttempts = attempts
token.RedeemNextRetryAt = &next
return nil
}
func (s *RockService) markRedeemingAbnormal(ctx context.Context, token *model.RockToken, reason string, recorder *RockDeliveryRecorder) {
func (s *RockService) markRedeemingAbnormal(ctx context.Context, token *model.RockToken, reason string, recorder *RockDeliveryRecorder) error {
fromStatus := token.Status
_ = s.db.WithContext(ctx).Unscoped().Model(token).Updates(map[string]interface{}{
if err := s.db.WithContext(ctx).Unscoped().Model(token).Updates(map[string]interface{}{
"status": model.RockStatusAbnormal,
"redeem_last_error": reason,
"redeem_next_retry_at": nil,
"redeem_timeout_warned_at": nil,
"remark": appendRockRemark(token.Remark, reason),
"updated_at": time.Now(),
}).Error
}).Error; err != nil {
return err
}
writeRockTokenStatusChange(s.db.WithContext(ctx), token, fromStatus, model.RockStatusAbnormal,
rockTokenSourceRedeem, reason, false, nil)
token.Status = model.RockStatusAbnormal
@@ -314,6 +434,7 @@ func (s *RockService) markRedeemingAbnormal(ctx context.Context, token *model.Ro
"to_status": model.RockStatusAbnormal,
})
}
return nil
}
func (s *RockService) syncDeliveredOrder(ctx context.Context, token *model.RockToken, recorder *RockDeliveryRecorder) {
+42 -3
View File
@@ -36,7 +36,7 @@ func TestCheckReleaseToken_UpstreamStatuses(t *testing.T) {
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"token": map[string]interface{}{
"status": "redeemed",
"redeemed_at": "2026-11-07 22:44:00",
"redeemed_at": time.Now().In(timeutil.Location()).Add(-10 * time.Second).Format("2006-01-02 15:04:05"),
},
})
case "abnormal":
@@ -179,9 +179,9 @@ func TestReleaseUnboundUsingToken_ForceClearsFieldsAndWritesLogs(t *testing.T) {
MismatchAt: &now,
MismatchReason: "历史不一致",
ClaimedAt: &claimedAt,
UsingAt: &now,
UsingAt: &claimedAt,
BoundAt: &now,
RedeemedAt: &now,
RedeemedAt: nil,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
@@ -251,6 +251,45 @@ func TestReleaseUnboundUsingToken_ForceClearsFieldsAndWritesLogs(t *testing.T) {
}
}
func TestReleaseUnboundUsingToken_ForceStillRejectsRedeemed(t *testing.T) {
db := newRockTestDB(t)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"token": map[string]interface{}{
"status": "redeemed",
"redeemed_at": time.Now().In(timeutil.Location()).Add(-time.Minute).Format("2006-01-02 15:04:05"),
},
})
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearer(t, svc, "bearer-test")
now := time.Now()
usingAt := now.Add(-time.Hour)
token := model.RockToken{
UpstreamTokenID: "release-redeemed",
Token: "jwt",
RedeemCode: "CODE-RELEASED",
Color: model.RockColorBlack,
Status: model.RockStatusUsing,
UsingAt: &usingAt,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
if err := svc.ReleaseUnboundUsingToken(context.Background(), token.ID, true, nil); err == nil {
t.Fatal("上游已发货时强制释放也必须被拒绝")
}
var stored model.RockToken
if err := db.First(&stored, token.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if stored.Status != model.RockStatusUsing {
t.Fatalf("拒绝释放后状态不应变化,实际 %s", stored.Status)
}
}
// newBoundReleaseServer 返回一个「上游已扫码绑定」的 mock,用于强制释放场景。
func newBoundReleaseServer(t *testing.T) *httptest.Server {
t.Helper()
@@ -229,8 +229,8 @@ func TestResetMismatchToken_ClearsRestockPriority(t *testing.T) {
}
}
// TestUpdateTokenRemark_ManualAvailableClearsRestockPriority 验证人工备注改回可用不产生优先权。
func TestUpdateTokenRemark_ManualAvailableClearsRestockPriority(t *testing.T) {
// TestUpdateTokenRemark_RejectsManualAvailable 防止备注接口绕过释放校验。
func TestUpdateTokenRemark_RejectsManualAvailable(t *testing.T) {
db := newRockTestDB(t)
svc := NewRockService(db, NewRockClient("https://example.com"))
ctx := context.Background()
@@ -248,19 +248,19 @@ func TestUpdateTokenRemark_ManualAvailableClearsRestockPriority(t *testing.T) {
t.Fatalf("创建 abnormal Token 失败: %v", err)
}
if _, err := svc.UpdateTokenRemark(ctx, token.ID, "人工改回可用", model.RockStatusAvailable, nil); err != nil {
t.Fatalf("人工改回可用失败: %v", err)
if _, err := svc.UpdateTokenRemark(ctx, token.ID, "人工改回可用", model.RockStatusAvailable, nil); err == nil {
t.Fatal("备注接口不应允许直接改回 available")
}
var updated model.RockToken
if err := db.First(&updated, token.ID).Error; err != nil {
t.Fatalf("读取更新结果失败: %v", err)
}
if updated.Status != model.RockStatusAvailable {
t.Fatalf("人工更新后应为 available,实际 %s", updated.Status)
if updated.Status != model.RockStatusAbnormal {
t.Fatalf("拒绝后应保持 abnormal,实际 %s", updated.Status)
}
if updated.RestockedAt != nil {
t.Fatalf("人工改回可用不应产生优先权,restocked_at=%v", updated.RestockedAt)
if updated.RestockedAt == nil {
t.Fatalf("拒绝后不应清空原有 restocked_at")
}
}
+22 -9
View File
@@ -53,11 +53,13 @@ type RockStatsResponse struct {
UsingDangerStock int64 `json:"using_danger_stock"`
BoundOverdueStock int64 `json:"bound_overdue_stock"`
// 异常标记计数(派生筛选,不是状态):历史流转中出现过异常的 Token 数。
AnomalyStock int64 `json:"anomaly_stock"`
UpstreamPoints float64 `json:"upstream_points"`
AutoDeliveryEnabled bool `json:"auto_delivery_enabled"`
TokenWarningStock int `json:"token_warning_stock"`
AutoPurchaseEnabled bool `json:"auto_purchase_enabled"`
AnomalyStock int64 `json:"anomaly_stock"`
NoEntitlementStock int64 `json:"no_entitlement_stock"`
UnresolvedAnomalyStock int64 `json:"unresolved_anomaly_stock"`
UpstreamPoints float64 `json:"upstream_points"`
AutoDeliveryEnabled bool `json:"auto_delivery_enabled"`
TokenWarningStock int `json:"token_warning_stock"`
AutoPurchaseEnabled bool `json:"auto_purchase_enabled"`
// 自动刷新绑定错误开关状态,供筛选栏 Switch 回显
AutoRefreshMismatchEnabled bool `json:"auto_refresh_mismatch_enabled"`
// 自动补货冷却状态:供页面提示“下次可补货时间”
@@ -223,7 +225,7 @@ func (s *RockService) UpdateConfig(ctx context.Context, req UpdateRockConfigReq)
return nil, &RockConfigValidationError{Message: "使用中红色预警天数必须大于橙色预警天数"}
}
// 绑定过期预警:已绑定超过 N 天视为绑定超时,独立于使用中两级预警。
// 绑定后未发货预警:本轮真实绑定后超过 N 天仍未发货。
boundWarnDays := cfg.BoundWarnDays
if req.BoundWarnDays != nil {
boundWarnDays = *req.BoundWarnDays
@@ -359,8 +361,6 @@ func (s *RockService) GetStats(ctx context.Context) (*RockStatsResponse, error)
for _, r := range results {
switch r.Status {
case model.RockStatusAvailable:
stats.AvailableStock = r.Total
case model.RockStatusUsing:
stats.UsingStock = r.Total
case model.RockStatusBound:
@@ -373,6 +373,12 @@ func (s *RockService) GetStats(ctx context.Context) (*RockStatsResponse, error)
stats.AbnormalStock = r.Total
}
}
// 实际分配只取黑色通用 Token,看板可用库存也使用同一口径。
if err := s.db.WithContext(ctx).Model(&model.RockToken{}).
Where("status = ? AND color = ?", model.RockStatusAvailable, model.RockColorBlack).
Count(&stats.AvailableStock).Error; err != nil {
return nil, err
}
var deletedCount int64
if err := s.db.WithContext(ctx).Unscoped().Model(&model.RockToken{}).
@@ -397,7 +403,8 @@ func (s *RockService) GetStats(ctx context.Context) (*RockStatsResponse, error)
return nil, err
}
if err := s.db.WithContext(ctx).Model(&model.RockToken{}).
Where("status = ? AND bound_at IS NOT NULL AND bound_at <= ?", model.RockStatusBound, rockOverdueCutoff(now, boundWarnDays)).
Where("status = ? AND bound_at IS NOT NULL AND bound_at <= ? AND redeem_requested_at IS NULL",
model.RockStatusBound, rockOverdueCutoff(now, boundWarnDays)).
Count(&stats.BoundOverdueStock).Error; err != nil {
return nil, err
}
@@ -406,6 +413,12 @@ func (s *RockService) GetStats(ctx context.Context) (*RockStatsResponse, error)
Count(&stats.AnomalyStock).Error; err != nil {
return nil, err
}
stats.UnresolvedAnomalyStock = stats.AnomalyStock
if err := rockTokenIncidentFilter(s.db.WithContext(ctx).Model(&model.RockToken{}),
model.RockIncidentCategoryNoEntitlement).
Count(&stats.NoEntitlementStock).Error; err != nil {
return nil, err
}
if cfg.AutoPurchaseEnabled {
if remain := rockAutoPurchaseCooldownRemaining(cfg, time.Now()); remain > 0 {
+133 -64
View File
@@ -9,56 +9,74 @@ import (
"affiliate_dash/internal/pkg/timeutil"
)
// rockBuyerNewlyBound 判断上游展示的绑定是不是本次领取之后新扫上的。
// 和分配快照一致,或者绑定时间仍为空且发货记录都早于本次领取,都是历史残留,链接可以再次使用。
// 没有快照、也没有更早的发货记录时,上游已有角色按新绑定处理,避免把刚扫上的号放回库存。
func rockBuyerNewlyBound(token *model.RockToken, ref *RockRefreshResponse) bool {
if ref == nil || !ref.IsBound() || strings.TrimSpace(ref.GetRoleID()) == "" {
return false
}
snap, hasSnap := bindingSnapshot(token)
if hasSnap && snap.Bound && snap.RoleID == ref.GetRoleID() && snap.BindChangedAt == strings.TrimSpace(ref.Token.BindChangedAt) {
return false
}
if deliveryAfterClaim(token, ref) || bindChangedAfterClaim(token, ref.Token.BindChangedAt) {
return true
}
if hasSnap && snap.RoleID != "" && snap.RoleID != ref.GetRoleID() {
return true
}
if strings.TrimSpace(ref.Token.BindChangedAt) == "" && historicalDeliveriesOnly(token, ref) {
return false
}
if hasSnap {
return false
}
return true
}
// rockUpstreamBound 判断上游「当前」是否已绑定。
//
// 只看本次实时返回的绑定信号(status / bind_changed_at / binding),
// 不用 bound_account_id 判断:上游可能保留上一次绑定的账号残留,
// 用它会导致刷新按钮把「未扫码」误判成已绑定,也无法把真实绑定同步回本地。
func rockUpstreamBound(ref *RockRefreshResponse) bool {
func rockBindingChangedAt(ref *RockRefreshResponse) string {
if ref == nil {
return false
return ""
}
switch strings.ToLower(strings.TrimSpace(ref.Token.Status)) {
case model.RockStatusBound, model.RockStatusRedeemed:
return true
if value := strings.TrimSpace(ref.Token.BindChangedAt); value != "" {
return value
}
if strings.TrimSpace(ref.Token.BindChangedAt) != "" {
return true
}
return ref.Binding.Bound || strings.TrimSpace(ref.Binding.RoleID) != ""
return strings.TrimSpace(ref.Binding.BindTime)
}
func historicalDeliveriesOnly(token *model.RockToken, ref *RockRefreshResponse) bool {
if ref == nil || len(ref.DeliveryList) == 0 {
// rockRawBindingAfterBaseline 只判断上游是否返回了本轮绑定事件,不参考残留账号。
func rockRawBindingAfterBaseline(token *model.RockToken, ref *RockRefreshResponse) (time.Time, bool) {
raw := rockBindingChangedAt(ref)
if raw == "" {
return time.Time{}, false
}
changedAt, ok := parseRockTime(raw)
if !ok {
return time.Time{}, false
}
baseline := rockClaimBaseline(token)
if baseline.IsZero() {
return time.Time{}, false
}
return changedAt, !changedAt.Before(baseline.Add(-time.Minute))
}
// rockCurrentCycleBinding 判断上游绑定事件是否属于当前分配周期。
func rockCurrentCycleBinding(token *model.RockToken, ref *RockRefreshResponse) (bool, time.Time) {
if token == nil || ref == nil {
return false, time.Time{}
}
changedAt, current := rockRawBindingAfterBaseline(token, ref)
if !current {
return false, time.Time{}
}
if rockBindingMatchesInitialSnapshot(token, ref) {
return false, time.Time{}
}
return true, changedAt
}
func rockBindingMatchesInitialSnapshot(token *model.RockToken, ref *RockRefreshResponse) bool {
snap, ok := bindingSnapshot(token)
if !ok {
return false
}
return !deliveryAfterClaim(token, ref)
return sameRockBindingEvent(snap.BindChangedAt, rockBindingChangedAt(ref))
}
// sameRockBindingEvent 兼容上游同一事件的不同时间格式,避免仅因字符串格式差异
// 把分配时保存的历史绑定重新识别成本轮绑定。
func sameRockBindingEvent(snapshotAt, upstreamAt string) bool {
snapshotAt = strings.TrimSpace(snapshotAt)
upstreamAt = strings.TrimSpace(upstreamAt)
if snapshotAt == "" || upstreamAt == "" {
return false
}
snapshotTime, snapshotOK := parseRockTime(snapshotAt)
upstreamTime, upstreamOK := parseRockTime(upstreamAt)
if snapshotOK && upstreamOK {
diff := snapshotTime.Sub(upstreamTime)
if diff < 0 {
diff = -diff
}
return diff <= time.Second
}
return snapshotAt == upstreamAt
}
func bindingSnapshot(token *model.RockToken) (BindingSnapshot, bool) {
@@ -78,10 +96,13 @@ func rockClaimBaseline(token *model.RockToken) time.Time {
if token == nil {
return time.Time{}
}
if token.UsingAt != nil && !token.UsingAt.IsZero() {
return token.UsingAt.UTC()
}
if token.ClaimedAt != nil && !token.ClaimedAt.IsZero() {
return token.ClaimedAt.UTC()
}
return token.CreatedAt.UTC()
return time.Time{}
}
// parseRockTime 解析上游返回的时间串。
@@ -106,34 +127,82 @@ func parseRockTime(raw string) (time.Time, bool) {
return time.Time{}, false
}
func bindChangedAfterClaim(token *model.RockToken, changedAt string) bool {
changed, ok := parseRockTime(changedAt)
if !ok {
return false
}
baseline := rockClaimBaseline(token)
if baseline.IsZero() {
return true
}
return !changed.Before(baseline.Add(-time.Minute))
type rockDeliveryEvidence struct {
Current bool
Ambiguous bool
At time.Time
FulfillID int64
Source string
}
func deliveryAfterClaim(token *model.RockToken, ref *RockRefreshResponse) bool {
if ref == nil {
return false
// rockCurrentCycleDelivery 只认本轮 using_at 之后产生的兑换时间或发货流水。
// 明确早于本轮的历史记录忽略;状态为 redeemed 却没有可解析时间时按
// “已发货、数据待核对”保守处理,禁止释放。
func rockCurrentCycleDelivery(token *model.RockToken, ref *RockRefreshResponse) rockDeliveryEvidence {
if token == nil || ref == nil {
return rockDeliveryEvidence{}
}
baseline := rockClaimBaseline(token)
if baseline.IsZero() {
return false
}
cutoff := baseline.Add(-time.Minute)
evidence := rockDeliveryEvidence{}
for _, item := range ref.DeliveryList {
deliveredAt, ok := parseRockTime(item.Time)
at, ok := parseRockTime(item.Time)
if !ok {
continue
}
if !deliveredAt.Before(baseline.Add(-time.Minute)) {
return true
if baseline.IsZero() {
if !evidence.Ambiguous || at.After(evidence.At) {
evidence.Ambiguous = true
evidence.At = at
evidence.FulfillID = item.FulfillID
evidence.Source = "delivery_list"
}
continue
}
if at.Before(cutoff) {
continue
}
if !evidence.Current || at.After(evidence.At) ||
(at.Equal(evidence.At) && item.FulfillID > evidence.FulfillID) {
evidence.Current = true
evidence.Ambiguous = false
evidence.At = at
evidence.FulfillID = item.FulfillID
evidence.Source = "delivery_list"
}
}
return false
rawRedeemedAt := strings.TrimSpace(ref.Token.RedeemedAt)
if rawRedeemedAt != "" {
if at, ok := parseRockTime(rawRedeemedAt); ok {
if baseline.IsZero() {
evidence.Ambiguous = true
if evidence.At.IsZero() || at.After(evidence.At) {
evidence.At = at
evidence.Source = "redeemed_at"
}
} else if !at.Before(cutoff) &&
(!evidence.Current || at.After(evidence.At)) {
evidence.Current = true
evidence.Ambiguous = false
evidence.At = at
evidence.Source = "redeemed_at"
}
} else {
evidence.Ambiguous = true
if evidence.Source == "" {
evidence.Source = "redeemed_at"
}
}
}
if strings.EqualFold(strings.TrimSpace(ref.Token.Status), model.RockStatusRedeemed) &&
rawRedeemedAt == "" && !evidence.Current {
evidence.Ambiguous = true
if evidence.Source == "" {
evidence.Source = "status"
}
}
return evidence
}
+2 -2
View File
@@ -34,7 +34,7 @@ type RockStaleRefreshResult struct {
// RefreshStaleTokens 定时巡检入口:受「自动刷新」开关控制,关闭时直接跳过。
//
// 与手动「一键刷新」使用同一套扫描规则:绑定错误 + 使用超时 + 绑定超时。
// 与手动「一键刷新」使用同一套扫描规则:绑定错误 + 待绑定超时 + 绑定后未发货。
func (s *RockService) RefreshStaleTokens(ctx context.Context) {
cfg, err := s.GetConfig(ctx)
if err != nil {
@@ -59,7 +59,7 @@ func (s *RockService) RefreshStaleTokens(ctx context.Context) {
result.Scanned, result.Fixed, result.Remaining, result.Failed)
}
// RefreshAllStaleTokens 手动「一键刷新」:绑定错误 + 使用超时 + 绑定超时,
// RefreshAllStaleTokens 手动「一键刷新」:绑定错误 + 待绑定超时 + 绑定后未发货,
// 忽略自动开关,同步执行并返回结果。
func (s *RockService) RefreshAllStaleTokens(ctx context.Context) (*RockStaleRefreshResult, error) {
return s.refreshStaleTokens(ctx)
+4
View File
@@ -567,6 +567,9 @@ func (s *RockService) ListTokens(ctx context.Context, page, pageSize int, status
if pageSize <= 0 {
pageSize = 20
}
if pageSize > 100 {
pageSize = 100
}
cfg, err := s.GetConfig(ctx)
if err != nil {
@@ -593,6 +596,7 @@ func (s *RockService) ListTokens(ctx context.Context, page, pageSize int, status
applyRockTokenOverdueFlags(tokens, now, warnDays, dangerDays, boundWarnDays)
s.prepareRockTokenList(ctx, tokens)
s.fillRockTokenAnomalyFlags(ctx, tokens)
s.fillRockTokenIncidentFlags(ctx, tokens)
return tokens, total, nil
}
+2
View File
@@ -55,6 +55,8 @@ func (s *RockService) ListSupportBoundTokens(ctx context.Context, page, pageSize
}
applyRockTokenOverdueFlags(tokens, now, warnDays, dangerDays, boundWarnDays)
s.prepareRockTokenList(ctx, tokens)
s.fillRockTokenAnomalyFlags(ctx, tokens)
s.fillRockTokenIncidentFlags(ctx, tokens)
return tokens, total, nil
}
+12 -5
View File
@@ -18,8 +18,9 @@ const (
rockTokenDeleteAllMax = 5000
// 虚拟筛选状态:不落库,只在查询时换算成「主状态 + 时间阈值」。
rockTokenStatusUsingOverdue = "using_overdue"
rockTokenStatusBoundOverdue = "bound_overdue"
rockTokenStatusUsingOverdue = "using_overdue"
rockTokenStatusBoundOverdue = "bound_overdue"
rockTokenStatusNoEntitlement = "no_entitlement"
)
var rockTokenDangerousStatuses = map[string]bool{
@@ -86,7 +87,8 @@ func isKnownRockTokenStatus(status string) bool {
switch status {
case "", model.RockStatusAvailable, model.RockStatusUsing, model.RockStatusBound,
model.RockStatusRedeemed, model.RockStatusMismatch, model.RockStatusAbnormal,
rockTokenStatusUsingOverdue, rockTokenStatusBoundOverdue, model.RockStatusAnomaly:
rockTokenStatusUsingOverdue, rockTokenStatusBoundOverdue, rockTokenStatusNoEntitlement,
model.RockStatusAnomaly:
return true
default:
return false
@@ -144,6 +146,9 @@ func applyRockTokenOverdueFlags(tokens []model.RockToken, now time.Time, warnDay
if tokens[i].BoundAt == nil {
continue
}
if rockTokenRedeemPending(&tokens[i]) {
continue
}
days := now.Sub(*tokens[i].BoundAt).Hours() / 24
if days < 0 {
days = 0
@@ -162,10 +167,12 @@ func applyRockTokenListFilters(db *gorm.DB, status, search string, now time.Time
db = db.Where("status = ? AND using_at IS NOT NULL AND using_at <= ?",
model.RockStatusUsing, rockOverdueCutoff(now, warnDays))
case rockTokenStatusBoundOverdue:
db = db.Where("status = ? AND bound_at IS NOT NULL AND bound_at <= ?",
db = db.Where("status = ? AND bound_at IS NOT NULL AND bound_at <= ? AND redeem_requested_at IS NULL",
model.RockStatusBound, rockOverdueCutoff(now, boundWarnDays))
case rockTokenStatusNoEntitlement:
db = rockTokenIncidentFilter(db, model.RockIncidentCategoryNoEntitlement)
case model.RockStatusAnomaly:
// 异常是派生筛选:只看历史流转,不限定主状态,也不改状态。
// 异常是派生筛选:包含历史异常和除“无权益”外的开放发货错误。
db = rockTokenAnomalyFilter(db)
default:
if value := strings.TrimSpace(status); value != "" {
@@ -82,6 +82,7 @@ func (s *RockService) ListDeletedRockTokens(ctx context.Context, page, pageSize
}
applyRockTokenOverdueFlags(tokens, now, warnDays, dangerDays, boundWarnDays)
s.fillRockTokenAnomalyFlags(ctx, tokens)
s.fillRockTokenIncidentFlags(ctx, tokens)
return tokens, total, nil
}
@@ -15,6 +15,7 @@ const (
rockTokenEventStatusChanged = "status_changed"
rockTokenEventDeleted = "deleted"
rockTokenEventRestored = "restored"
rockTokenEventIncident = "incident"
)
// Token 状态来源,用于前端展示流转原因。
@@ -31,6 +32,7 @@ const (
rockTokenSourcePackageSync = "package_sync"
rockTokenSourceDelete = "delete"
rockTokenSourceRestore = "restore"
rockTokenSourceIncident = "incident"
)
// writeRockTokenStatusLog 记录 Token 生命周期/状态变化。
+5 -11
View File
@@ -7,10 +7,8 @@ import (
"net/http/httptest"
"strings"
"testing"
"time"
"affiliate_dash/internal/model"
"affiliate_dash/internal/pkg/timeutil"
)
func TestAssignTokenForOrder_SetsUsingAtAndReleaseClears(t *testing.T) {
@@ -89,7 +87,7 @@ func TestAssignTokenForOrder_SetsUsingAtAndReleaseClears(t *testing.T) {
}
}
func TestImportPackageDetail_SetsUsingAtFromClaimedAt(t *testing.T) {
func TestImportPackageDetail_ResidualUsingStatusBecomesAvailable(t *testing.T) {
db := newRockTestDB(t)
svc := NewRockService(db, NewRockClient("https://example.com"))
@@ -116,15 +114,11 @@ func TestImportPackageDetail_SetsUsingAtFromClaimedAt(t *testing.T) {
if err := db.Where("upstream_token_id = ?", "30001").First(&row).Error; err != nil {
t.Fatalf("查询导入 Token 失败: %v", err)
}
if row.Status != model.RockStatusUsing {
t.Fatalf("期望状态 using,实际 %s", row.Status)
if row.Status != model.RockStatusAvailable {
t.Fatalf("无真实绑定事件时应导入为 available,实际 %s", row.Status)
}
if row.UsingAt == nil {
t.Fatalf("上游状态为 using 时必须写入 using_at")
}
expected := time.Date(2026, 10, 27, 23, 31, 59, 0, timeutil.Location())
if !row.UsingAt.Equal(expected) {
t.Fatalf("using_at 应取上游 claimed_at,实际 %v", row.UsingAt)
if row.UsingAt != nil {
t.Fatalf("available 记录不应写入 using_at,实际 %v", row.UsingAt)
}
}