feat: Rock 兑换支持上游异步 pending 并记录发货全链路报文

问题:上游 0666.pro 兑换返回 HTTP 200 + status=pending(官方处理中),
旧逻辑仅凭「没有 error 字段」就判定成功并写 redeemed 终态,导致本地显示
已发货而上游仍是 bound,且 redeemed 为硬终态无法自愈。

后端:
- RedeemSkin 改为三态判定(success / pending / failed),不再无 error 即成功
- 新增 redeeming「发货处理中」状态;pending 不写 redeemed_at/fulfill_id
- 成功落库前二次 refresh 确认上游 redeemed 或 redeemed_at 才置终态
- 新增确认 worker:20s 轮询、30s→5m 指数退避、110001 全局冷却 60s、
  15 分钟未确认仅告警不回退,确认后同步订单为 delivered 并触发回调
- pending 期间禁止重复兑换,订单保持 delivering,不下发发货成功回调
- 新增 rock_delivery_logs 表与记录器:入站参数、内部状态、上游请求/响应
  原文全部留档,保留 180 天
- 分配/绑定校验/刷新/导入/删除/回收站/统计/客服查询全部适配 redeeming
- 迁移 026:redeeming 跟踪字段、部分索引与报文日志表

前端:
- Token 列表新增「发货处理中」筛选项、标签与超时提示
- 删除/回收站纳入 redeeming 危险状态
- 历史抽屉拆为「状态历史 / 发货报文」双页签,按 trace 展示报文详情
- 买家发货页在 pending 时提示官方处理中,无需重复提交

测试:新增 pending 不落终态、异常响应不误判、重复提交不重复兑换、
确认后同步订单、保持 pending 并退避 5 个用例;go build/vet/test 与
前端构建均通过。
This commit is contained in:
2026-11-14 12:09:16 +08:00
parent 227d4b93ad
commit f0dea6a583
33 changed files with 2241 additions and 170 deletions
@@ -0,0 +1,54 @@
-- 026: Rock 异步发货(上游 pending)状态 + 发货全链路报文日志
ALTER TABLE rock_tokens ADD COLUMN IF NOT EXISTS redeem_requested_at timestamptz;
ALTER TABLE rock_tokens ADD COLUMN IF NOT EXISTS redeem_checked_at timestamptz;
ALTER TABLE rock_tokens ADD COLUMN IF NOT EXISTS redeem_attempts integer NOT NULL DEFAULT 0;
ALTER TABLE rock_tokens ADD COLUMN IF NOT EXISTS redeem_next_retry_at timestamptz;
ALTER TABLE rock_tokens ADD COLUMN IF NOT EXISTS redeem_last_message text NOT NULL DEFAULT '';
ALTER TABLE rock_tokens ADD COLUMN IF NOT EXISTS redeem_last_error text NOT NULL DEFAULT '';
ALTER TABLE rock_tokens ADD COLUMN IF NOT EXISTS redeem_timeout_warned_at timestamptz;
COMMENT ON COLUMN rock_tokens.redeem_requested_at IS '首次提交上游兑换的时间(pending 起点,15 分钟未确认触发告警)';
COMMENT ON COLUMN rock_tokens.redeem_checked_at IS '最近一次查询上游兑换结果的时间';
COMMENT ON COLUMN rock_tokens.redeem_attempts IS '已执行的兑换结果轮询次数';
COMMENT ON COLUMN rock_tokens.redeem_next_retry_at IS '下一次允许轮询上游的时间(指数退避)';
COMMENT ON COLUMN rock_tokens.redeem_last_message IS '上游最近一次返回的处理说明';
COMMENT ON COLUMN rock_tokens.redeem_last_error IS '最近一次轮询/兑换错误';
COMMENT ON COLUMN rock_tokens.redeem_timeout_warned_at IS '超过 15 分钟仍未确认时写入告警时间';
CREATE INDEX IF NOT EXISTS idx_rock_tokens_redeem_next_retry_at
ON rock_tokens (redeem_next_retry_at)
WHERE status = 'redeeming';
CREATE TABLE IF NOT EXISTS rock_delivery_logs (
id BIGSERIAL PRIMARY KEY,
trace_id VARCHAR(64) NOT NULL DEFAULT '',
token_id BIGINT NOT NULL DEFAULT 0,
upstream_token_id VARCHAR(64) NOT NULL DEFAULT '',
order_no VARCHAR(128) NOT NULL DEFAULT '',
step VARCHAR(40) NOT NULL DEFAULT '',
direction VARCHAR(16) NOT NULL DEFAULT '',
status VARCHAR(16) NOT NULL DEFAULT '',
http_method VARCHAR(8) NOT NULL DEFAULT '',
http_url TEXT NOT NULL DEFAULT '',
http_status INTEGER NOT NULL DEFAULT 0,
request_headers TEXT NOT NULL DEFAULT '',
request_body TEXT NOT NULL DEFAULT '',
response_headers TEXT NOT NULL DEFAULT '',
response_body TEXT NOT NULL DEFAULT '',
duration_ms INTEGER NOT NULL DEFAULT 0,
error_message TEXT NOT NULL DEFAULT '',
actor_user_id BIGINT,
actor_ip VARCHAR(64) NOT NULL DEFAULT '',
user_agent VARCHAR(512) NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
finished_at TIMESTAMPTZ
);
CREATE INDEX IF NOT EXISTS idx_rock_delivery_logs_trace
ON rock_delivery_logs (trace_id);
CREATE INDEX IF NOT EXISTS idx_rock_delivery_logs_token
ON rock_delivery_logs (token_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_rock_delivery_logs_order
ON rock_delivery_logs (order_no, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_rock_delivery_logs_created_at
ON rock_delivery_logs (created_at DESC);
+15
View File
@@ -478,6 +478,21 @@ func (h *RockHandler) ListTokenStatusLogs(c *gin.Context) {
response.OK(c, gin.H{"list": logs})
}
// ListDeliveryLogs 查询 Token 发货全链路报文(本系统入站 / 上游请求 / 上游响应)。
func (h *RockHandler) ListDeliveryLogs(c *gin.Context) {
idUint, err := strconv.ParseUint(c.Param("id"), 10, 64)
if err != nil {
response.BadRequest(c, "无效的 Token ID")
return
}
logs, err := h.rockSvc.ListRockDeliveryLogs(c.Request.Context(), uint(idUint))
if err != nil {
response.BadRequest(c, err.Error())
return
}
response.OK(c, gin.H{"list": logs})
}
// ReleaseToken 把使用中的提货链接释放回可用库存;上游异常时需人工强制确认。
func (h *RockHandler) ReleaseToken(c *gin.Context) {
idUint, err := strconv.ParseUint(c.Param("id"), 10, 64)
+76 -35
View File
@@ -23,6 +23,7 @@ const (
RockStatusAvailable = "available" // 可用库存
RockStatusUsing = "using" // 已分配给订单,正在提货扫码
RockStatusBound = "bound" // 扫码绑定成功
RockStatusRedeeming = "redeeming" // 上游官方异步处理中,等待确认发货结果
RockStatusRedeemed = "redeemed" // 发货完成
RockStatusMismatch = "mismatch" // UID 校验不符阻断(绑错)
RockStatusAbnormal = "abnormal" // 异常/作废
@@ -73,41 +74,50 @@ func (RockConfig) TableName() string {
// RockToken 通用 Token 本地库存池记录
type RockToken struct {
ID uint `gorm:"primaryKey" json:"id"`
UpstreamTokenID string `gorm:"size:64;not null;default:''" json:"upstream_token_id"`
Token string `gorm:"type:text;not null" json:"-"`
RedeemCode string `gorm:"size:64;not null;index" json:"redeem_code"`
Color string `gorm:"size:32;not null;default:'black';index" json:"color"`
Region string `gorm:"size:64;not null;default:'Q/V可用'" json:"region"`
Status string `gorm:"size:32;not null;default:'available';index" json:"status"`
PackageID *int64 `gorm:"index" json:"package_id,omitempty"`
PackageShareCode string `gorm:"size:64;not null;default:''" json:"package_share_code"`
AssignedOrderNo string `gorm:"size:128;index" json:"assigned_order_no"`
ClientOrderNo string `gorm:"-" json:"client_order_no,omitempty"` // 关联的创建订单商户单号
TokenCode string `gorm:"-" json:"token,omitempty"` // 提取码,仅管理端响应展示,不持久化
ExpectedUID string `gorm:"size:64" json:"expected_uid"`
ActualRoleName string `gorm:"size:128" json:"actual_role_name"`
ActualRoleID string `gorm:"size:64" json:"actual_role_id"`
ActualChannel string `gorm:"size:64" json:"actual_channel"`
InitialBoundSnapshot string `gorm:"type:text" json:"-"` // 分配时初始旧数据快照 JSON
UpstreamFulfillID int64 `json:"upstream_fulfill_id"`
TargetSkin string `gorm:"size:128" json:"target_skin"`
TargetCurrency string `gorm:"size:64;not null;default:'幸运币'" json:"target_currency"`
TargetCurrencyQuantity int `gorm:"not null;default:0" json:"target_currency_quantity"`
MismatchAt *time.Time `gorm:"index" json:"mismatch_at"`
MismatchReason string `gorm:"size:255" json:"mismatch_reason"`
Remark string `gorm:"type:text;not null;default:''" json:"remark"`
RawData string `gorm:"type:text;not null;default:''" json:"raw_data,omitempty"`
Source string `gorm:"size:32;not null;default:'purchase';index" json:"source"`
ClaimedAt *time.Time `json:"claimed_at"`
UsingAt *time.Time `gorm:"index" json:"using_at"` // 变为 using 的时间,超时预警起点
RestockedAt *time.Time `gorm:"index" json:"restocked_at,omitempty"` // 人工释放回库存的时间,分配时优先(FIFO)
BoundAt *time.Time `json:"bound_at"`
RedeemedAt *time.Time `json:"redeemed_at"`
DeletedAt gorm.DeletedAt `gorm:"index" json:"-"`
DeletedAtText string `gorm:"-" json:"deleted_at,omitempty"` // 回收站展示用,仅管理端响应返回
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
ID uint `gorm:"primaryKey" json:"id"`
UpstreamTokenID string `gorm:"size:64;not null;default:''" json:"upstream_token_id"`
Token string `gorm:"type:text;not null" json:"-"`
RedeemCode string `gorm:"size:64;not null;index" json:"redeem_code"`
Color string `gorm:"size:32;not null;default:'black';index" json:"color"`
Region string `gorm:"size:64;not null;default:'Q/V可用'" json:"region"`
Status string `gorm:"size:32;not null;default:'available';index" json:"status"`
PackageID *int64 `gorm:"index" json:"package_id,omitempty"`
PackageShareCode string `gorm:"size:64;not null;default:''" json:"package_share_code"`
AssignedOrderNo string `gorm:"size:128;index" json:"assigned_order_no"`
ClientOrderNo string `gorm:"-" json:"client_order_no,omitempty"` // 关联的创建订单商户单号
TokenCode string `gorm:"-" json:"token,omitempty"` // 提取码,仅管理端响应展示,不持久化
ExpectedUID string `gorm:"size:64" json:"expected_uid"`
ActualRoleName string `gorm:"size:128" json:"actual_role_name"`
ActualRoleID string `gorm:"size:64" json:"actual_role_id"`
ActualChannel string `gorm:"size:64" json:"actual_channel"`
InitialBoundSnapshot string `gorm:"type:text" json:"-"` // 分配时初始旧数据快照 JSON
UpstreamFulfillID int64 `json:"upstream_fulfill_id"`
TargetSkin string `gorm:"size:128" json:"target_skin"`
TargetCurrency string `gorm:"size:64;not null;default:'幸运币'" json:"target_currency"`
TargetCurrencyQuantity int `gorm:"not null;default:0" json:"target_currency_quantity"`
MismatchAt *time.Time `gorm:"index" json:"mismatch_at"`
MismatchReason string `gorm:"size:255" json:"mismatch_reason"`
Remark string `gorm:"type:text;not null;default:''" json:"remark"`
RawData string `gorm:"type:text;not null;default:''" json:"raw_data,omitempty"`
Source string `gorm:"size:32;not null;default:'purchase';index" json:"source"`
ClaimedAt *time.Time `json:"claimed_at"`
UsingAt *time.Time `gorm:"index" json:"using_at"` // 变为 using 的时间,超时预警起点
RestockedAt *time.Time `gorm:"index" json:"restocked_at,omitempty"` // 人工释放回库存的时间,分配时优先(FIFO)
BoundAt *time.Time `json:"bound_at"`
RedeemedAt *time.Time `json:"redeemed_at"`
// 异步兑换(上游 pending)跟踪字段。
RedeemRequestedAt *time.Time `gorm:"index" json:"redeem_requested_at,omitempty"` // 首次提交兑换的时间,用于 15 分钟超时告警
RedeemCheckedAt *time.Time `json:"redeem_checked_at,omitempty"` // 最近一次查询上游结果的时间
RedeemAttempts int `gorm:"not null;default:0" json:"redeem_attempts"`
RedeemNextRetryAt *time.Time `gorm:"index" json:"redeem_next_retry_at,omitempty"`
RedeemLastMessage string `gorm:"type:text;not null;default:''" json:"redeem_last_message,omitempty"`
RedeemLastError string `gorm:"type:text;not null;default:''" json:"redeem_last_error,omitempty"`
RedeemTimeoutWarnedAt *time.Time `json:"redeem_timeout_warned_at,omitempty"`
DeletedAt gorm.DeletedAt `gorm:"index" json:"-"`
DeletedAtText string `gorm:"-" json:"deleted_at,omitempty"` // 回收站展示用,仅管理端响应返回
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
// 以下为列表派生字段,不落库:由后端用与筛选/统计完全相同的 now 计算。
BoundOverdue bool `gorm:"-" json:"bound_overdue"`
@@ -140,6 +150,37 @@ func (RockTokenStatusLog) TableName() string {
return "rock_token_status_logs"
}
// RockDeliveryLog 记录一次发货链路的完整报文(本系统入站 / 上游请求 / 上游响应 / 内部状态)。
// 用于订单排障与对账;按 180 天保留,由后台任务定期清理。
type RockDeliveryLog struct {
ID uint `gorm:"primaryKey" json:"id"`
TraceID string `gorm:"size:64;not null;default:'';index:idx_rock_delivery_logs_trace" json:"trace_id"`
TokenID uint `gorm:"not null;default:0;index:idx_rock_delivery_logs_token" json:"token_id"`
UpstreamTokenID string `gorm:"size:64;not null;default:''" json:"upstream_token_id"`
OrderNo string `gorm:"size:128;not null;default:'';index:idx_rock_delivery_logs_order" json:"order_no"`
Step string `gorm:"size:40;not null;default:'';index" json:"step"`
Direction string `gorm:"size:16;not null;default:''" json:"direction"` // inbound / outbound / internal
Status string `gorm:"size:16;not null;default:'';index" json:"status"` // started / success / pending / failed / skipped
HTTPMethod string `gorm:"size:8;not null;default:''" json:"http_method"`
HTTPURL string `gorm:"type:text;not null;default:''" json:"http_url"`
HTTPStatus int `gorm:"not null;default:0" json:"http_status"`
RequestHeaders string `gorm:"type:text;not null;default:''" json:"request_headers"`
RequestBody string `gorm:"type:text;not null;default:''" json:"request_body"`
ResponseHeaders string `gorm:"type:text;not null;default:''" json:"response_headers"`
ResponseBody string `gorm:"type:text;not null;default:''" json:"response_body"`
DurationMs int `gorm:"not null;default:0" json:"duration_ms"`
ErrorMessage string `gorm:"type:text;not null;default:''" json:"error_message"`
ActorUserID *uint `gorm:"index" json:"actor_user_id,omitempty"`
ActorIP string `gorm:"size:64;not null;default:''" json:"actor_ip"`
UserAgent string `gorm:"size:512;not null;default:''" json:"user_agent"`
CreatedAt time.Time `json:"created_at"`
FinishedAt *time.Time `json:"finished_at,omitempty"`
}
func (RockDeliveryLog) TableName() string {
return "rock_delivery_logs"
}
// RockPurchaseLog 是一次采购尝试的主记录。
// 每一次真正进入采购流程的尝试都会落一条,包含降级原因与各阶段汇总。
type RockPurchaseLog struct {
+1
View File
@@ -300,6 +300,7 @@ func Setup(h *Handlers) *gin.Engine {
platform.POST("/platform/rock/tokens/:id/release", h.Rock.ReleaseToken)
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.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)
@@ -30,16 +30,38 @@ func buildExistingDeliverySubmitResult(order *model.FulfillmentOrder) *DeliveryS
status := normalizeOrderStatus(order)
if status == model.OrderStatusDelivered {
message = "订单已交付"
} else if boolFromMap(resultData, "rock_redeem_pending") {
message = "官方处理中,请稍后刷新查看结果(系统会自动确认发货状态)"
}
return &DeliverySubmitResult{
OrderNo: order.OrderNo,
Status: status,
Pending: boolFromMap(resultData, "rock_redeem_pending"),
Message: message,
ProviderOrderNo: providerOrderNo,
UpstreamOrder: resultData,
}
}
func boolFromMap(values map[string]interface{}, key string) bool {
if values == nil {
return false
}
switch value := values[key].(type) {
case bool:
return value
case string:
return strings.EqualFold(strings.TrimSpace(value), "true") || strings.TrimSpace(value) == "1"
case float64:
return value != 0
case json.Number:
parsed, err := value.Int64()
return err == nil && parsed != 0
default:
return false
}
}
func upstreamDeliverySucceeded(order map[string]interface{}) bool {
return strings.EqualFold(stringFromMap(order, "status"), "FINISHED") ||
strings.EqualFold(stringFromMap(order, "send_status"), "SUCCESS")
@@ -3,6 +3,7 @@ package service
import (
"context"
"errors"
"fmt"
"net/http"
"strings"
"time"
@@ -121,6 +122,7 @@ func (s *DeliveryService) submit(orderNo, gameAccount, bindUUID string, apiClien
return &DeliverySubmitResult{
OrderNo: updated.OrderNo,
Status: normalizeOrderStatus(updated),
Pending: submitRes.Pending,
Message: submitRes.Message,
ProviderOrderNo: submitRes.ProviderOrderNo,
GameAccount: submitRes.GameAccount,
@@ -218,6 +220,47 @@ func (s *DeliveryService) recordDeliverySubmissionStage(order *model.Fulfillment
return nil
}
// ConfirmRockDelivery 在 Rock 后台确认上游已发货后,把订单同步为已交付并触发回调。
func (s *DeliveryService) ConfirmRockDelivery(ctx context.Context, order *model.FulfillmentOrder, token *model.RockToken) error {
if order == nil || token == nil {
return errors.New("订单或 Token 不存在")
}
fulfillIDStr := ""
if token.UpstreamFulfillID > 0 {
fulfillIDStr = fmt.Sprintf("%d", token.UpstreamFulfillID)
}
skinName := token.TargetSkin
if strings.TrimSpace(skinName) == "" {
skinName = model.RockSkinBlackTshirt
}
currencyName := token.TargetCurrency
if strings.TrimSpace(currencyName) == "" {
currencyName = "幸运币"
}
patch := map[string]interface{}{
"provider": ProviderRock,
"provider_order_no": fulfillIDStr,
"fulfill_id": token.UpstreamFulfillID,
"rock_redeem_pending": false,
"rock_token_id": token.ID,
"rock_redeem_code": token.RedeemCode,
"skin": skinName,
"currency": currencyName,
"currency_quantity": token.TargetCurrencyQuantity,
"role_name": token.ActualRoleName,
"game_channel": token.ActualChannel,
"game_uid": token.ActualRoleID,
}
_, err := s.fulfillment.UpdateFulfillment(FulfillmentUpdateInput{
MerchantID: order.MerchantID,
OrderNo: order.OrderNo,
Status: model.OrderStatusDelivered,
ProviderOrderNo: fulfillIDStr,
ResultData: mergeResultData(order.ResultData, patch),
})
return err
}
func (s *DeliveryService) markDeliverySubmissionFailed(order *model.FulfillmentOrder, apiClientID uint, stage, reason string) {
if order == nil {
return
@@ -56,6 +56,7 @@ type DeliveryLinkResult struct {
type DeliverySubmitResult struct {
OrderNo string `json:"order_no"`
Status string `json:"status"`
Pending bool `json:"pending,omitempty"`
Message string `json:"message"`
ProviderOrderNo string `json:"provider_order_no,omitempty"`
GameAccount map[string]interface{} `json:"game_account,omitempty"`
+1
View File
@@ -118,6 +118,7 @@ type ProviderBindResult struct {
type ProviderSubmitResult struct {
ProviderOrderNo string `json:"provider_order_no"`
NextStatus string `json:"next_status"` // delivering / delivered
Pending bool `json:"pending,omitempty"`
Message string `json:"message"`
GameAccount map[string]interface{} `json:"game_account,omitempty"`
UpstreamOrder interface{} `json:"delivery,omitempty"`
+87 -27
View File
@@ -61,6 +61,7 @@ func (p *RockProvider) PrepareDelivery(ctx context.Context, order *model.Fulfill
Where("assigned_order_no = ? AND status IN ?", orderNo, []string{
model.RockStatusUsing,
model.RockStatusBound,
model.RockStatusRedeeming,
model.RockStatusMismatch,
model.RockStatusAbnormal,
}).
@@ -247,26 +248,25 @@ func (p *RockProvider) Submit(ctx context.Context, order *model.FulfillmentOrder
expectedUID = resultDataString(order.ResultData, "expected_uid")
}
redeemResp, err := p.rockService.RedeemTokenForOrder(ctx, order.OrderNo, sku, expectedUID)
recorder := p.rockService.NewDeliveryRecorder(order.OrderNo, nil, "", "")
recorder.Internal(ctx, "submit_request", "started", "订单进入 Rock 发货流程", map[string]interface{}{
"order_no": order.OrderNo,
"product_sku": sku,
"expected_uid": expectedUID,
"bind_uuid": bindUUID,
"provider": ProviderRock,
})
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{}{
"order_no": order.OrderNo,
})
return nil, err
}
var token model.RockToken
_ = p.rockService.db.WithContext(ctx).Unscoped().Where("assigned_order_no = ?", order.OrderNo).Order("id DESC").First(&token).Error
fulfillIDStr := ""
if token.UpstreamFulfillID > 0 {
fulfillIDStr = fmt.Sprintf("%d", token.UpstreamFulfillID)
}
skinName := redeemResp.Token.RedeemSkin
if skinName == "" {
skinName = token.TargetSkin
}
currencyName := redeemResp.Token.RedeemCurrency
if currencyName == "" {
currencyName = token.TargetCurrency
token := result.Token
if token == nil {
return nil, errors.New("发货结果缺少 Token 信息")
}
roleName := token.ActualRoleName
@@ -278,18 +278,78 @@ func (p *RockProvider) Submit(ctx context.Context, order *model.FulfillmentOrder
channel = resultDataString(order.ResultData, "game_channel")
}
patch := map[string]interface{}{
"provider": ProviderRock,
"provider_order_no": fulfillIDStr,
"fulfill_id": token.UpstreamFulfillID,
"skin": skinName,
"currency": currencyName,
"currency_quantity": token.TargetCurrencyQuantity,
"role_name": roleName,
"game_channel": channel,
"game_uid": expectedUID,
// 上游仍在异步处理:订单保持 delivering,不触发「已发货」回调。
if result.Pending {
recorder.Internal(ctx, "order_update", "pending", result.Message, map[string]interface{}{
"order_no": order.OrderNo,
"next_status": model.OrderStatusDelivering,
"token_id": token.ID,
"pending_since": token.RedeemRequestedAt,
})
return &ProviderSubmitResult{
NextStatus: model.OrderStatusDelivering,
Pending: true,
Message: "官方处理中,请稍后刷新查看结果(系统会自动确认发货状态)",
PatchResultData: map[string]interface{}{
"provider": ProviderRock,
"provider_order_stage": "redeem_pending",
"rock_redeem_pending": true,
"rock_token_id": token.ID,
"rock_redeem_code": token.RedeemCode,
"expected_uid": expectedUID,
"role_name": roleName,
"game_channel": channel,
"game_uid": expectedUID,
"skin": token.TargetSkin,
"currency": token.TargetCurrency,
"currency_quantity": token.TargetCurrencyQuantity,
},
}, nil
}
fulfillIDStr := ""
if token.UpstreamFulfillID > 0 {
fulfillIDStr = fmt.Sprintf("%d", token.UpstreamFulfillID)
}
skinName := token.TargetSkin
currencyName := token.TargetCurrency
if result.Response != nil {
if name := strings.TrimSpace(result.Response.Token.RedeemSkin); name != "" {
skinName = name
}
if name := strings.TrimSpace(result.Response.Token.RedeemCurrency); name != "" {
currencyName = name
}
}
if skinName == "" {
skinName = model.RockSkinBlackTshirt
}
if currencyName == "" {
currencyName = "幸运币"
}
patch := map[string]interface{}{
"provider": ProviderRock,
"provider_order_no": fulfillIDStr,
"fulfill_id": token.UpstreamFulfillID,
"rock_redeem_pending": false,
"rock_token_id": token.ID,
"rock_redeem_code": token.RedeemCode,
"skin": skinName,
"currency": currencyName,
"currency_quantity": token.TargetCurrencyQuantity,
"role_name": roleName,
"game_channel": channel,
"game_uid": expectedUID,
}
recorder.Internal(ctx, "order_update", "success", "订单发货完成", map[string]interface{}{
"order_no": order.OrderNo,
"next_status": model.OrderStatusDelivered,
"upstream_fulfill": fulfillIDStr,
})
return &ProviderSubmitResult{
ProviderOrderNo: fulfillIDStr,
NextStatus: model.OrderStatusDelivered,
@@ -303,9 +303,13 @@ func TestRockProvider_SubmitBlueAndBlack(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
if strings.HasSuffix(r.URL.Path, "/refresh") {
// 真实上游在礼品发放完成后会把 token.status 置为 redeemed 并写入 redeemed_at;
// 本地只有在拿到这两个信号后才允许落 redeemed 终态。
var resp RockRefreshResponse
resp.Token.Status = model.RockStatusRedeemed
resp.Token.RedeemedAt = "2026-11-14 11:00:00"
resp.DeliveryList = []RockDeliveryItem{
{FulfillID: 888999},
{FulfillID: 888999, Time: "2026-11-14 11:00:00"},
}
_ = json.NewEncoder(w).Encode(resp)
return
+3
View File
@@ -45,6 +45,7 @@ func (s *RockService) AssignTokenForOrder(ctx context.Context, orderNo, expected
Where("assigned_order_no = ? AND status IN ?", orderNo, []string{
model.RockStatusUsing,
model.RockStatusBound,
model.RockStatusRedeeming,
model.RockStatusMismatch,
model.RockStatusRedeemed,
model.RockStatusAbnormal,
@@ -60,6 +61,8 @@ func (s *RockService) AssignTokenForOrder(ctx context.Context, orderNo, expected
switch token.Status {
case model.RockStatusAbnormal:
return nil, nil, errors.New(rockOrderTokenLockedMessage)
case model.RockStatusRedeeming:
return nil, nil, errors.New("该订单已提交发货,上游官方处理中,不能再次领取提货号")
case model.RockStatusRedeemed:
return nil, nil, errors.New("该订单已发货,不能再次领取提货号")
case model.RockStatusMismatch:
@@ -83,6 +83,18 @@ func (s *RockService) CheckBindingAndVerifyUID(ctx context.Context, tokenID uint
}, nil
}
// 已提交发货(上游异步处理中)的 Token 不再参与扫码绑定轮询。
if token.Status == model.RockStatusRedeeming {
return &CheckBindingResult{
Bound: true,
RoleName: token.ActualRoleName,
RoleID: token.ActualRoleID,
Channel: token.ActualChannel,
ExpectedUID: token.ExpectedUID,
Message: "订单已提交发货,上游官方处理中",
}, nil
}
// 已释放回库存、或已改派给其他订单的链接,不能再被旧发货页轮询改写。
if token.Status != model.RockStatusUsing && token.Status != model.RockStatusBound {
return &CheckBindingResult{
@@ -27,7 +27,7 @@ type TokenBindingRefreshResult struct {
func canRefreshRockBinding(status string) bool {
switch status {
case model.RockStatusUsing, model.RockStatusBound, model.RockStatusMismatch:
case model.RockStatusUsing, model.RockStatusBound, model.RockStatusRedeeming, model.RockStatusMismatch:
return true
default:
return false
@@ -169,6 +169,17 @@ func (s *RockService) refreshTokenBinding(ctx context.Context, tokenID uint, act
case upstreamBound:
// 上游已绑定即同步为 bound:不以本地预期 UID 做拦截,
// 否则历史 UID 不一致的 Token 永远无法把真实状态刷新回本地。
// 但本地已进入 redeeming 时不能降级回 bound,必须保持发货处理中。
if token.Status == model.RockStatusRedeeming {
updates["status"] = model.RockStatusRedeeming
updates["redeem_checked_at"] = now
if msg := strings.TrimSpace(ref.Message); msg != "" {
updates["redeem_last_message"] = msg
}
result.Status = model.RockStatusRedeeming
result.Message = "上游仍未确认发货,保持发货处理中"
break
}
updates["status"] = model.RockStatusBound
updates["mismatch_at"] = nil
updates["mismatch_reason"] = ""
@@ -187,7 +198,12 @@ func (s *RockService) refreshTokenBinding(ctx context.Context, tokenID uint, act
}
default:
result.Message = "上游仍未绑定,本地状态保持不变"
if token.Status == model.RockStatusRedeeming {
updates["redeem_checked_at"] = now
result.Message = "上游仍未确认发货,保持发货处理中"
} else {
result.Message = "上游仍未绑定,本地状态保持不变"
}
}
result.Changed = result.Status != originalStatus ||
+150 -23
View File
@@ -257,12 +257,47 @@ func (r *RockRefreshResponse) GetChannel() string {
type RockRedeemResponse struct {
Token RockTokenData `json:"token"`
Status string `json:"status"`
Pending bool `json:"pending"`
Message string `json:"message"`
Error string `json:"error,omitempty"`
}
func (r *RockRedeemResponse) IsSuccess() bool {
return r.Status == "success" || r.Token.Status == "redeemed"
if r == nil {
return false
}
return strings.EqualFold(strings.TrimSpace(r.Status), "success") ||
strings.EqualFold(strings.TrimSpace(r.Token.Status), model.RockStatusRedeemed) ||
strings.TrimSpace(r.Token.RedeemedAt) != ""
}
// IsPending 表示上游已受理兑换,但腾讯官方仍在异步处理,需要稍后刷新确认。
// 上游 0666.pro 的返回示例:
//
// {"token":{"status":"bound",...},"status":"pending","pending":true,
// "message":"官方处理中,请稍后刷新查看结果"}
func (r *RockRedeemResponse) IsPending() bool {
if r == nil {
return false
}
return r.Pending || strings.EqualFold(strings.TrimSpace(r.Status), "pending")
}
// FailureMessage 提取上游失败原因,供报错与日志使用。
func (r *RockRedeemResponse) FailureMessage(fallback string) string {
if r == nil {
return fallback
}
if msg := strings.TrimSpace(r.Error); msg != "" {
return msg
}
if msg := strings.TrimSpace(r.Message); msg != "" {
return msg
}
if status := strings.TrimSpace(r.Status); status != "" {
return "上游返回状态: " + status
}
return fallback
}
type RockExchangeRecordsResponse struct {
@@ -836,25 +871,51 @@ func (c *RockClient) RefreshToken(ctx context.Context, bearerToken string, upstr
req.Header.Set("Origin", c.BaseURL())
req.Header.Set("Referer", c.BaseURL()+"/")
resp, err := c.httpClient.Do(req)
recorder := rockDeliveryRecorderFrom(ctx)
var logEntry *model.RockDeliveryLog
startedAt := time.Now()
if recorder != nil {
logEntry = recorder.StartUpstream(ctx, "refresh_token", http.MethodPost, reqURL, req.Header, nil)
}
var resp *http.Response
var body []byte
finish := func(status string, callErr error) {
if recorder != nil {
recorder.FinishUpstream(ctx, logEntry, status, resp, body, callErr, time.Since(startedAt))
}
}
resp, err = c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("请求 0666 refresh 失败: %w", err)
wrapped := fmt.Errorf("请求 0666 refresh 失败: %w", err)
finish("failed", wrapped)
return nil, wrapped
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
body, err = io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("读取 refresh 响应失败: %w", err)
wrapped := fmt.Errorf("读取 refresh 响应失败: %w", err)
finish("failed", wrapped)
return nil, wrapped
}
if resp.StatusCode != http.StatusOK {
return nil, &RockUpstreamStatusError{StatusCode: resp.StatusCode, Body: string(body)}
if isRockRateLimitError(fmt.Errorf("%s", string(body))) {
markRockUpstreamRateLimited()
}
wrapped := &RockUpstreamStatusError{StatusCode: resp.StatusCode, Body: string(body)}
finish("failed", wrapped)
return nil, wrapped
}
var res RockRefreshResponse
if err := json.Unmarshal(body, &res); err != nil {
return nil, fmt.Errorf("解析 refresh 响应失败: %w", err)
wrapped := fmt.Errorf("解析 refresh 响应失败: %w", err)
finish("failed", wrapped)
return nil, wrapped
}
finish("success", nil)
return &res, nil
}
@@ -883,28 +944,58 @@ func (c *RockClient) RefreshRedeemBinding(ctx context.Context, redeemCode string
req.Header.Set("Origin", c.BaseURL())
req.Header.Set("Referer", c.BaseURL()+"/")
resp, err := c.httpClient.Do(req)
recorder := rockDeliveryRecorderFrom(ctx)
var logEntry *model.RockDeliveryLog
startedAt := time.Now()
if recorder != nil {
logEntry = recorder.StartUpstream(ctx, "refresh_redeem_binding", http.MethodPost, reqURL, req.Header, nil)
}
var resp *http.Response
var body []byte
finish := func(status string, callErr error) {
if recorder != nil {
recorder.FinishUpstream(ctx, logEntry, status, resp, body, callErr, time.Since(startedAt))
}
}
resp, err = c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("请求 0666 redeem refresh 失败: %w", err)
wrapped := fmt.Errorf("请求 0666 redeem refresh 失败: %w", err)
finish("failed", wrapped)
return nil, wrapped
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
body, err = io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("读取 redeem refresh 响应失败: %w", err)
wrapped := fmt.Errorf("读取 redeem refresh 响应失败: %w", err)
finish("failed", wrapped)
return nil, wrapped
}
if resp.StatusCode != http.StatusOK {
return nil, &RockUpstreamStatusError{StatusCode: resp.StatusCode, Body: string(body)}
if isRockRateLimitError(fmt.Errorf("%s", string(body))) {
markRockUpstreamRateLimited()
}
wrapped := &RockUpstreamStatusError{StatusCode: resp.StatusCode, Body: string(body)}
finish("failed", wrapped)
return nil, wrapped
}
var res RockRefreshResponse
if err := json.Unmarshal(body, &res); err != nil {
return nil, fmt.Errorf("解析 redeem refresh 响应失败: %w", err)
wrapped := fmt.Errorf("解析 redeem refresh 响应失败: %w", err)
finish("failed", wrapped)
return nil, wrapped
}
finish("success", nil)
return &res, nil
}
// RedeemSkin 执行最终皮肤兑换发货(需携带商户 Bearer Token 鉴权)
// RedeemSkin 执行最终皮肤兑换发货(需携带商户 Bearer Token 鉴权)。
//
// 上游是异步处理:可能返回 status=success(已完成),也可能返回
// status=pending / pending=true(腾讯官方处理中)。两种都返回 nil error,
// 由调用方区分处理;其他情况一律视为失败,避免把业务失败误判成发货成功。
func (c *RockClient) RedeemSkin(ctx context.Context, bearerToken, upstreamTokenID, skin, currency, account string) (*RockRedeemResponse, error) {
reqURL := fmt.Sprintf("%s/api/redeem/%s", c.BaseURL(), url.PathEscape(upstreamTokenID))
payload := map[string]interface{}{
@@ -926,29 +1017,65 @@ func (c *RockClient) RedeemSkin(ctx context.Context, bearerToken, upstreamTokenI
req.Header.Set("Origin", c.BaseURL())
req.Header.Set("Referer", c.BaseURL()+"/")
resp, err := c.httpClient.Do(req)
recorder := rockDeliveryRecorderFrom(ctx)
var logEntry *model.RockDeliveryLog
startedAt := time.Now()
if recorder != nil {
logEntry = recorder.StartUpstream(ctx, "redeem_request", http.MethodPost, reqURL, req.Header, jsonBytes)
}
var resp *http.Response
var body []byte
finish := func(status string, callErr error) {
if recorder != nil {
recorder.FinishUpstream(ctx, logEntry, status, resp, body, callErr, time.Since(startedAt))
}
}
resp, err = c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("请求 0666 兑换发货接口失败: %w", err)
wrapped := fmt.Errorf("请求 0666 兑换发货接口失败: %w", err)
finish("failed", wrapped)
return nil, wrapped
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
body, err = io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("读取兑换响应失败: %w", err)
wrapped := fmt.Errorf("读取兑换响应失败: %w", err)
finish("failed", wrapped)
return nil, wrapped
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("上游兑换发货接口返回错误码 %d: %s", resp.StatusCode, string(body))
if isRockRateLimitError(fmt.Errorf("%s", string(body))) {
markRockUpstreamRateLimited()
}
wrapped := fmt.Errorf("上游兑换发货接口返回错误码 %d: %s", resp.StatusCode, string(body))
finish("failed", wrapped)
return nil, wrapped
}
var res RockRedeemResponse
if err := json.Unmarshal(body, &res); err != nil {
return nil, fmt.Errorf("解析兑换响应失败: %w", err)
wrapped := fmt.Errorf("解析兑换响应失败: %w", err)
finish("failed", wrapped)
return nil, wrapped
}
if !res.IsSuccess() && res.Error != "" {
return nil, fmt.Errorf("上游兑换失败: %s", res.Error)
// 异步受理:上游已收到请求,官方仍在处理中。
if res.IsPending() {
finish("pending", nil)
return &res, nil
}
return &res, nil
// 明确成功:必须能证明已完成兑换。
if res.IsSuccess() {
finish("success", nil)
return &res, nil
}
wrapped := fmt.Errorf("上游兑换未成功: %s", res.FailureMessage(strings.TrimSpace(string(body))))
finish("failed", wrapped)
return nil, wrapped
}
// QueryExchangeRecords 异步查询兑换对账记录
@@ -0,0 +1,280 @@
package service
import (
"context"
"encoding/json"
"fmt"
"log"
"net/http"
"strings"
"sync/atomic"
"time"
"affiliate_dash/internal/model"
"gorm.io/gorm"
)
const (
// 发货报文日志保留天数:按业务要求原样明文保存 180 天。
rockDeliveryLogRetentionDays = 180
// 单个报文字段最大长度,避免异常大响应拖垮数据库;超出部分截断并标注。
rockDeliveryLogMaxBodyBytes = 512 * 1024
// 上游 110001 频控后的全局冷却时间。
rockRateLimitCooldown = 60 * time.Second
)
// rockUpstreamRateLimitedUntil 记录上游频控解禁时间(UnixNano)。
// 买家扫码轮询与后台确认任务共用,避免把 0666.pro 打成持续 110001。
var rockUpstreamRateLimitedUntil atomic.Int64
func markRockUpstreamRateLimited() {
rockUpstreamRateLimitedUntil.Store(time.Now().Add(rockRateLimitCooldown).UnixNano())
}
func rockUpstreamRateLimitActive() bool {
until := rockUpstreamRateLimitedUntil.Load()
return until > 0 && time.Now().UnixNano() < until
}
func isRockRateLimitError(err error) bool {
if err == nil {
return false
}
msg := err.Error()
return strings.Contains(msg, "110001") || strings.Contains(msg, "请求过于频繁")
}
// RockDeliveryRecorder 负责把一次发货链路的每一步写入 rock_delivery_logs。
//
// 设计要点:
// - 上游调用采用「预写日志」:请求前先落 started,拿到响应后更新为最终状态,
// 即使进程中途重启也能看到未完成的调用;
// - 入站请求(本系统收到的参数)与内部状态迁移也各落一条,前端按 trace_id 串起来;
// - 写入为 best-effort,失败只打日志,绝不阻断发货主流程。
type RockDeliveryRecorder struct {
db *gorm.DB
TraceID string
TokenID uint
UpstreamTokenID string
OrderNo string
ActorUserID *uint
ActorIP string
UserAgent string
}
// NewDeliveryRecorder 创建一次发货链路的记录器。
func (s *RockService) NewDeliveryRecorder(orderNo string, actorUserID *uint, actorIP, userAgent string) *RockDeliveryRecorder {
return &RockDeliveryRecorder{
db: s.db,
TraceID: fmt.Sprintf("rd-%d-%d", time.Now().UnixNano(), time.Now().UnixNano()%100000),
OrderNo: strings.TrimSpace(orderNo),
ActorUserID: actorUserID,
ActorIP: strings.TrimSpace(actorIP),
UserAgent: strings.TrimSpace(userAgent),
}
}
// AttachToken 把 Token 身份补到后续日志上(Token 在订单内查询后才能确定)。
func (r *RockDeliveryRecorder) AttachToken(token *model.RockToken) {
if r == nil || token == nil {
return
}
r.TokenID = token.ID
r.UpstreamTokenID = strings.TrimSpace(token.UpstreamTokenID)
}
// Inbound 记录本系统收到的一次业务请求(买家提交发货等)。
func (r *RockDeliveryRecorder) Inbound(ctx context.Context, step string, payload interface{}) {
r.write(ctx, &model.RockDeliveryLog{
TraceID: r.traceID(),
TokenID: r.TokenID,
UpstreamTokenID: r.UpstreamTokenID,
OrderNo: r.OrderNo,
Step: step,
Direction: "inbound",
Status: "success",
RequestHeaders: "",
RequestBody: rockJSONText(payload),
ResponseBody: "",
ActorUserID: r.ActorUserID,
ActorIP: r.ActorIP,
UserAgent: r.UserAgent,
})
}
// Internal 记录一个内部步骤(状态迁移、订单更新、回调等)。
func (r *RockDeliveryRecorder) Internal(ctx context.Context, step, status, message string, payload interface{}) {
if r == nil {
return
}
entry := &model.RockDeliveryLog{
TraceID: r.traceID(),
TokenID: r.TokenID,
UpstreamTokenID: r.UpstreamTokenID,
OrderNo: r.OrderNo,
Step: step,
Direction: "internal",
Status: status,
RequestBody: rockJSONText(payload),
ErrorMessage: strings.TrimSpace(message),
ActorUserID: r.ActorUserID,
ActorIP: r.ActorIP,
UserAgent: r.UserAgent,
}
now := time.Now()
entry.FinishedAt = &now
r.write(ctx, entry)
}
// StartUpstream 在调用上游前预写一条 started 日志,返回的 entry 用于回填响应。
func (r *RockDeliveryRecorder) StartUpstream(ctx context.Context, step, method, reqURL string, headers http.Header, reqBody []byte) *model.RockDeliveryLog {
if r == nil {
return nil
}
entry := &model.RockDeliveryLog{
TraceID: r.traceID(),
TokenID: r.TokenID,
UpstreamTokenID: r.UpstreamTokenID,
OrderNo: r.OrderNo,
Step: step,
Direction: "outbound",
Status: "started",
HTTPMethod: method,
HTTPURL: reqURL,
RequestHeaders: rockHeadersText(headers),
RequestBody: truncateRockDeliveryBody(reqBody),
ActorUserID: r.ActorUserID,
ActorIP: r.ActorIP,
UserAgent: r.UserAgent,
}
if err := r.db.WithContext(ctx).Create(entry).Error; err != nil {
log.Printf("[Rock] 写入发货请求日志失败 trace=%s step=%s: %v", r.TraceID, step, err)
return nil
}
return entry
}
// FinishUpstream 回填上游响应报文与耗时。
func (r *RockDeliveryRecorder) FinishUpstream(ctx context.Context, entry *model.RockDeliveryLog, status string, resp *http.Response, respBody []byte, callErr error, duration time.Duration) {
if r == nil || entry == nil {
return
}
updates := map[string]interface{}{
"status": status,
"duration_ms": int(duration.Milliseconds()),
"response_body": truncateRockDeliveryBody(respBody),
"finished_at": time.Now(),
}
if resp != nil {
updates["http_status"] = resp.StatusCode
updates["response_headers"] = rockHeadersText(resp.Header)
}
if callErr != nil {
updates["error_message"] = callErr.Error()
}
if err := r.db.WithContext(ctx).Model(&model.RockDeliveryLog{}).
Where("id = ?", entry.ID).
Updates(updates).Error; err != nil {
log.Printf("[Rock] 回填发货响应日志失败 trace=%s id=%d: %v", r.TraceID, entry.ID, err)
}
}
func (r *RockDeliveryRecorder) traceID() string {
if r == nil || strings.TrimSpace(r.TraceID) == "" {
return "rd-unknown"
}
return r.TraceID
}
func (r *RockDeliveryRecorder) write(ctx context.Context, entry *model.RockDeliveryLog) {
if r == nil || r.db == nil || entry == nil {
return
}
entry.TraceID = r.traceID()
if entry.TokenID == 0 {
entry.TokenID = r.TokenID
}
if entry.UpstreamTokenID == "" {
entry.UpstreamTokenID = r.UpstreamTokenID
}
if entry.OrderNo == "" {
entry.OrderNo = r.OrderNo
}
if err := r.db.WithContext(ctx).Create(entry).Error; err != nil {
log.Printf("[Rock] 写入发货链路日志失败 trace=%s step=%s: %v", entry.TraceID, entry.Step, err)
}
}
type rockDeliveryRecorderCtxKey struct{}
func withRockDeliveryRecorder(ctx context.Context, recorder *RockDeliveryRecorder) context.Context {
if recorder == nil {
return ctx
}
return context.WithValue(ctx, rockDeliveryRecorderCtxKey{}, recorder)
}
func rockDeliveryRecorderFrom(ctx context.Context) *RockDeliveryRecorder {
if ctx == nil {
return nil
}
if recorder, ok := ctx.Value(rockDeliveryRecorderCtxKey{}).(*RockDeliveryRecorder); ok {
return recorder
}
return nil
}
// ListRockDeliveryLogs 返回指定 Token 的完整发货链路(按时间正序,便于前端时间线展示)。
func (s *RockService) ListRockDeliveryLogs(ctx context.Context, tokenID uint) ([]model.RockDeliveryLog, error) {
var logs []model.RockDeliveryLog
if err := s.db.WithContext(ctx).
Where("token_id = ?", tokenID).
Order("id ASC").
Limit(500).
Find(&logs).Error; err != nil {
return nil, err
}
return logs, nil
}
// CleanupRockDeliveryLogs 清理超过保留期的发货报文日志。
func (s *RockService) CleanupRockDeliveryLogs(ctx context.Context) {
cutoff := time.Now().AddDate(0, 0, -rockDeliveryLogRetentionDays)
if err := s.db.WithContext(ctx).
Where("created_at < ?", cutoff).
Delete(&model.RockDeliveryLog{}).Error; err != nil {
log.Printf("[Rock] 清理发货报文日志失败: %v", err)
}
}
func rockHeadersText(headers http.Header) string {
if len(headers) == 0 {
return ""
}
encoded, err := json.Marshal(headers)
if err != nil {
return ""
}
return string(encoded)
}
func rockJSONText(value interface{}) string {
if value == nil {
return ""
}
encoded, err := json.Marshal(value)
if err != nil {
return fmt.Sprintf("%v", value)
}
return string(encoded)
}
func truncateRockDeliveryBody(body []byte) string {
if len(body) == 0 {
return ""
}
if len(body) <= rockDeliveryLogMaxBodyBytes {
return string(body)
}
return string(body[:rockDeliveryLogMaxBodyBytes]) + "\n...[truncated]"
}
+4 -1
View File
@@ -123,7 +123,8 @@ func (s *RockService) importPackageDetail(ctx context.Context, detail *RockPacka
oldStatus := existing.Status
canRefreshUpstream := existing.Source == "package" &&
strings.TrimSpace(existing.AssignedOrderNo) == "" &&
existing.Status != model.RockStatusAvailable
existing.Status != model.RockStatusAvailable &&
existing.Status != model.RockStatusRedeeming
if canRefreshUpstream {
updates["status"] = status
updates["restocked_at"] = nil // 上游包同步回库不属于「释放回库存」,不产生优先权
@@ -181,6 +182,8 @@ func normalizeRockPackageStatus(status string) string {
return model.RockStatusUsing
case model.RockStatusBound:
return model.RockStatusBound
case model.RockStatusRedeeming:
return model.RockStatusRedeeming
case model.RockStatusRedeemed:
return model.RockStatusRedeemed
case model.RockStatusMismatch:
+3
View File
@@ -160,6 +160,9 @@ func (s *RockService) UpdateTokenRemark(ctx context.Context, tokenID uint, remar
trimmedStatus := strings.TrimSpace(nextStatus)
if trimmedStatus != "" {
if token.Status == model.RockStatusRedeeming && trimmedStatus != model.RockStatusAbnormal {
return nil, fmt.Errorf("发货处理中的 Token 只能标记为已上报失效,不能改回其他状态")
}
switch trimmedStatus {
case model.RockStatusAbnormal, model.RockStatusAvailable, model.RockStatusMismatch:
updates["status"] = trimmedStatus
+228 -15
View File
@@ -14,8 +14,20 @@ import (
"gorm.io/gorm/clause"
)
// RedeemTokenForOrder 执行最终发货兑换(动态判定皮肤)
func (s *RockService) RedeemTokenForOrder(ctx context.Context, orderNo, sku, expectedUID string) (*RockRedeemResponse, error) {
// RockRedeemResult 是一次自动发货尝试的结果。
type RockRedeemResult struct {
Token *model.RockToken
Response *RockRedeemResponse
Pending bool
Message string
TraceID string
}
// RedeemTokenForOrder 执行最终发货兑换(动态判定皮肤)。
//
// 上游 0666.pro 是异步兑奖:HTTP 200 + status=pending 表示「官方处理中」,
// 此时绝不能写 redeemed 终态,而是进入 redeeming 并交给后台任务轮询确认。
func (s *RockService) RedeemTokenForOrder(ctx context.Context, orderNo, sku, expectedUID string, recorder *RockDeliveryRecorder) (*RockRedeemResult, error) {
cfg, err := s.GetConfig(ctx)
if err != nil {
return nil, err
@@ -24,9 +36,10 @@ func (s *RockService) RedeemTokenForOrder(ctx context.Context, orderNo, sku, exp
return nil, fmt.Errorf("Rock 自动发货通道维护中,已关闭自动兑换")
}
// 已进入 redeeming 的 Token 不重复提交兑换,直接返回处理中,避免重复发放。
var token model.RockToken
err = s.db.WithContext(ctx).Unscoped().
Where("assigned_order_no = ? AND status = ?", orderNo, model.RockStatusBound).
Where("assigned_order_no = ? AND status IN ?", orderNo, []string{model.RockStatusBound, model.RockStatusRedeeming}).
First(&token).Error
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
@@ -34,6 +47,27 @@ func (s *RockService) RedeemTokenForOrder(ctx context.Context, orderNo, sku, exp
}
return nil, err
}
if recorder != nil {
recorder.AttachToken(&token)
}
ctx = withRockDeliveryRecorder(ctx, recorder)
if token.Status == model.RockStatusRedeeming {
msg := strings.TrimSpace(token.RedeemLastMessage)
if msg == "" {
msg = "上游官方处理中,请稍后刷新查看结果"
}
if recorder != nil {
recorder.Internal(ctx, "submit_request", "pending", msg, map[string]interface{}{
"order_no": orderNo,
"token_id": token.ID,
"upstream_token_id": token.UpstreamTokenID,
"expected_uid": expectedUID,
"duplicated": true,
})
}
return &RockRedeemResult{Token: &token, Pending: true, Message: msg, TraceID: recorderTraceID(recorder)}, nil
}
// 最终双重防御:核验 UID
trimmedExpected := strings.TrimSpace(expectedUID)
@@ -50,48 +84,227 @@ func (s *RockService) RedeemTokenForOrder(ctx context.Context, orderNo, sku, exp
targetSkin = model.RockSkinBlueSportswear
}
// 确保买家角色名存在
// 确保买家角色名存在。上游要求 account 必须与绑定角色名一致,
// 不能用 UID 兜底,否则会被上游拒绝兑换。
actualRoleName := strings.TrimSpace(token.ActualRoleName)
if actualRoleName == "" {
if ref, refErr := s.client.RefreshToken(ctx, cfg.BearerToken, token.UpstreamTokenID); refErr == nil && ref != nil {
actualRoleName = ref.GetRoleName()
actualRoleName = strings.TrimSpace(ref.GetRoleName())
if actualRoleName != "" {
_ = s.db.WithContext(ctx).Unscoped().Model(&token).Update("actual_role_name", actualRoleName).Error
token.ActualRoleName = actualRoleName
}
if token.ActualRoleID == "" && strings.TrimSpace(ref.GetRoleID()) != "" {
token.ActualRoleID = strings.TrimSpace(ref.GetRoleID())
}
}
}
if actualRoleName == "" {
actualRoleName = token.ActualRoleID
return nil, fmt.Errorf("上游未返回绑定角色名,无法兑换发货,请刷新绑定状态后重试")
}
if recorder != nil {
recorder.Internal(ctx, "submit_request", "started", "提交上游兑换", map[string]interface{}{
"order_no": orderNo,
"token_id": token.ID,
"upstream_token_id": token.UpstreamTokenID,
"expected_uid": trimmedExpected,
"actual_role_id": token.ActualRoleID,
"actual_role_name": actualRoleName,
"skin": targetSkin,
"currency": "幸运币",
})
}
// 调上游兑换发货接口(携带 Bearer Token 鉴权)
redeemResp, err := s.client.RedeemSkin(ctx, cfg.BearerToken, token.UpstreamTokenID, targetSkin, "幸运币", actualRoleName)
if err != nil {
if recorder != nil {
recorder.Internal(ctx, "redeem_result", "failed", err.Error(), nil)
}
if isRockRateLimitError(err) {
markRockUpstreamRateLimited()
}
return nil, fmt.Errorf("调用上游兑换失败: %w", err)
}
// 提取官方 fulfillId
var fulfillID int64
now := time.Now()
ref, refErr := s.client.RefreshToken(ctx, cfg.BearerToken, token.UpstreamTokenID)
if refErr == nil && len(ref.DeliveryList) > 0 {
fulfillID = ref.DeliveryList[len(ref.DeliveryList)-1].FulfillID
// 上游已受理但官方仍在异步处理:进入 redeeming,等待后台轮询确认。
if redeemResp.IsPending() {
msg := redeemResp.FailureMessage("官方处理中,请稍后刷新查看结果")
now := time.Now()
token = s.markTokenRedeeming(ctx, token, targetSkin, msg, recorder)
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
}
// 状态跃迁至 redeemed
// 上游返回成功后再 refresh 一次,确认 token 已真正进入 redeemed 且有履约流水。
ref, refErr := s.client.RefreshToken(ctx, cfg.BearerToken, token.UpstreamTokenID)
if refErr != nil {
if isRockRateLimitError(refErr) {
markRockUpstreamRateLimited()
}
if recorder != nil {
recorder.Internal(ctx, "redeem_confirm", "pending", "兑换成功但二次确认失败,等待后台确认: "+refErr.Error(), nil)
}
// 刷新失败不直接否认可信的成功响应,但也不立即写终态;交给后台轮询确认。
msg := "上游已受理,正在等待发货结果确认"
token = s.markTokenRedeeming(ctx, token, targetSkin, msg, recorder)
return &RockRedeemResult{Token: &token, Response: redeemResp, Pending: true, Message: msg, TraceID: recorderTraceID(recorder)}, nil
}
if ref != nil && !rockUpstreamRedeemed(ref) {
msg := "上游返回成功但尚未确认发货,正在等待官方结果"
if recorder != nil {
recorder.Internal(ctx, "redeem_confirm", "pending", msg, map[string]interface{}{
"upstream_status": ref.Token.Status,
"redeemed_at": ref.Token.RedeemedAt,
})
}
token = s.markTokenRedeeming(ctx, token, targetSkin, msg, recorder)
return &RockRedeemResult{Token: &token, Response: redeemResp, Pending: true, Message: msg, TraceID: recorderTraceID(recorder)}, nil
}
fulfillID := extractRockFulfillID(ref, rockClaimBaseline(&token))
redeemedAt := time.Now()
if ref != nil {
if parsed, ok := parseRockTime(ref.Token.RedeemedAt); ok {
redeemedAt = parsed
}
}
// 状态跃迁至 redeemed(只有拿到明确成功/明显履约结果才执行)
fromStatus := token.Status
now := time.Now()
_ = s.db.WithContext(ctx).Unscoped().Model(&token).Updates(map[string]interface{}{
"status": model.RockStatusRedeemed,
"upstream_fulfill_id": fulfillID,
"target_skin": targetSkin,
"target_currency": "幸运币",
"target_currency_quantity": 1,
"redeemed_at": now,
"redeemed_at": redeemedAt,
"redeem_requested_at": nil,
"redeem_checked_at": now,
"redeem_next_retry_at": nil,
"redeem_last_message": "",
"redeem_last_error": "",
"redeem_timeout_warned_at": nil,
}).Error
writeRockTokenStatusChange(s.db.WithContext(ctx), &token, fromStatus, model.RockStatusRedeemed,
rockTokenSourceRedeem, "自动发货: "+targetSkin, false, nil)
if recorder != nil {
recorder.Internal(ctx, "state_change", "success", "发货完成", map[string]interface{}{
"from_status": fromStatus,
"to_status": model.RockStatusRedeemed,
"upstream_fulfill_id": fulfillID,
"redeemed_at": redeemedAt.Format("2006-01-02 15:04:05"),
})
}
return redeemResp, nil
token.Status = model.RockStatusRedeemed
token.UpstreamFulfillID = fulfillID
token.TargetSkin = targetSkin
token.TargetCurrency = "幸运币"
token.TargetCurrencyQuantity = 1
token.RedeemedAt = &redeemedAt
return &RockRedeemResult{Token: &token, Response: redeemResp, Pending: false, Message: "发货成功", TraceID: recorderTraceID(recorder)}, nil
}
// markTokenRedeeming 把 Token 置为「发货处理中」,记录 pending 起点与首次轮询时间。
func (s *RockService) markTokenRedeeming(ctx context.Context, token model.RockToken, targetSkin, message string, recorder *RockDeliveryRecorder) model.RockToken {
now := time.Now()
requestedAt := token.RedeemRequestedAt
if requestedAt == nil {
requestedAt = &now
}
nextRetry := now.Add(30 * time.Second)
fromStatus := token.Status
_ = s.db.WithContext(ctx).Unscoped().Model(&token).Updates(map[string]interface{}{
"status": model.RockStatusRedeeming,
"target_skin": targetSkin,
"target_currency": "幸运币",
"redeem_requested_at": requestedAt,
"redeem_checked_at": now,
"redeem_attempts": 0,
"redeem_next_retry_at": nextRetry,
"redeem_last_message": strings.TrimSpace(message),
"redeem_last_error": "",
"updated_at": now,
}).Error
writeRockTokenStatusChange(s.db.WithContext(ctx), &token, fromStatus, model.RockStatusRedeeming,
rockTokenSourceRedeem, strings.TrimSpace(message), false, nil)
if recorder != nil {
recorder.Internal(ctx, "state_change", "pending", strings.TrimSpace(message), map[string]interface{}{
"from_status": fromStatus,
"to_status": model.RockStatusRedeeming,
"redeem_requested_at": requestedAt.Format("2006-01-02 15:04:05"),
"next_retry_at": nextRetry.Format("2006-01-02 15:04:05"),
})
}
token.Status = model.RockStatusRedeeming
token.TargetSkin = targetSkin
token.TargetCurrency = "幸运币"
token.RedeemRequestedAt = requestedAt
token.RedeemCheckedAt = &now
token.RedeemAttempts = 0
token.RedeemNextRetryAt = &nextRetry
token.RedeemLastMessage = strings.TrimSpace(message)
token.RedeemLastError = ""
return token
}
func recorderTraceID(recorder *RockDeliveryRecorder) string {
if recorder == nil {
return ""
}
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
}
var best int64
var bestAt time.Time
for _, item := range ref.DeliveryList {
if item.FulfillID <= 0 {
continue
}
at, ok := parseRockTime(item.Time)
if ok && !baseline.IsZero() && at.Before(baseline.Add(-time.Minute)) {
continue
}
if at.After(bestAt) || (at.Equal(bestAt) && item.FulfillID > best) {
best = item.FulfillID
bestAt = at
}
}
if best > 0 {
return best
}
for _, item := range ref.DeliveryList {
if item.FulfillID > best {
best = item.FulfillID
}
}
return best
}
// RockReleaseCheckResult 是释放回库存前的上游校验结果。
@@ -0,0 +1,365 @@
package service
import (
"context"
"errors"
"fmt"
"log"
"strings"
"sync/atomic"
"time"
"affiliate_dash/internal/model"
)
const (
rockRedeemPollInterval = 20 * time.Second
rockRedeemBatchSize = 20
rockRedeemCallGap = 2 * time.Second
rockRedeemTimeout = 15 * time.Minute
rockRedeemFirstRetry = 30 * time.Second
rockRedeemMaxRetry = 5 * time.Minute
)
var rockRedeemWorkerRunning atomic.Bool
// RockRedeemConfirmSummary 是一次 pending 确认批次的汇总。
type RockRedeemConfirmSummary struct {
Scanned int `json:"scanned"`
Confirmed int `json:"confirmed"`
Pending int `json:"pending"`
Failed int `json:"failed"`
TimedOut int `json:"timed_out"`
}
// SetRedeemConfirmHook 注册「上游确认发货后同步订单」的回调,由 main 装配 DeliveryService。
func (s *RockService) SetRedeemConfirmHook(hook func(ctx context.Context, order *model.FulfillmentOrder, token *model.RockToken) error) {
s.redeemConfirmHook = hook
}
// RunRedeemConfirmWorker 启动 pending 兑换确认任务:每 20 秒扫描一次 redeeming Token。
//
// 频控保护:上游 110001 时全局冷却 60 秒,且每个 Token 使用指数退避,
// 避免把 0666.pro 的公开刷新接口打成持续频控。
func (s *RockService) RunRedeemConfirmWorker(ctx context.Context) {
lastCleanup := time.Now()
ticker := time.NewTicker(rockRedeemPollInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if rockUpstreamRateLimitActive() {
continue
}
if _, err := s.ProcessRedeemingTokens(ctx); err != nil && !errors.Is(err, errRockRedeemBusy) {
log.Printf("[Rock] 发货结果确认任务失败: %v", err)
}
if time.Since(lastCleanup) >= 24*time.Hour {
s.CleanupRockDeliveryLogs(ctx)
lastCleanup = time.Now()
}
}
}
}
var errRockRedeemBusy = errors.New("发货结果确认任务正在执行")
// ProcessRedeemingTokens 扫描并确认一批 pending 兑换。可被后台任务或测试直接调用。
func (s *RockService) ProcessRedeemingTokens(ctx context.Context) (*RockRedeemConfirmSummary, error) {
if !rockRedeemWorkerRunning.CompareAndSwap(false, true) {
return nil, errRockRedeemBusy
}
defer rockRedeemWorkerRunning.Store(false)
now := time.Now()
var tokens []model.RockToken
if err := s.db.WithContext(ctx).
Where("status = ?", model.RockStatusRedeeming).
Where("(redeem_next_retry_at IS NULL OR redeem_next_retry_at <= ?)", now).
Order("redeem_next_retry_at ASC").
Limit(rockRedeemBatchSize).
Find(&tokens).Error; err != nil {
return nil, err
}
summary := &RockRedeemConfirmSummary{Scanned: len(tokens)}
for i := range tokens {
if i > 0 {
if rockUpstreamRateLimitActive() {
break
}
select {
case <-ctx.Done():
return summary, ctx.Err()
case <-time.After(rockRedeemCallGap):
}
}
confirmed, timedOut, err := s.confirmRedeemingToken(ctx, &tokens[i])
if err != nil {
summary.Failed++
log.Printf("[Rock] 确认 Token %d 发货结果失败: %v", tokens[i].ID, err)
continue
}
if confirmed {
summary.Confirmed++
continue
}
summary.Pending++
if timedOut {
summary.TimedOut++
}
}
return summary, nil
}
// confirmRedeemingToken 查询一次上游结果并更新本地状态。
func (s *RockService) confirmRedeemingToken(ctx context.Context, token *model.RockToken) (confirmed bool, timedOut bool, err error) {
if token == nil {
return false, false, errors.New("Token 不存在")
}
cfg, err := s.GetConfig(ctx)
if err != nil {
return false, false, err
}
recorder := s.NewDeliveryRecorder(token.AssignedOrderNo, nil, "", "")
recorder.AttachToken(token)
ctx = withRockDeliveryRecorder(ctx, recorder)
ref, err := s.refreshRedeemStatus(ctx, token, cfg.BearerToken)
if err != nil {
if isRockRateLimitError(err) {
markRockUpstreamRateLimited()
}
s.bumpRedeemRetry(ctx, token, err.Error(), false)
if recorder != nil {
recorder.Internal(ctx, "redeem_confirm", "failed", err.Error(), nil)
}
return false, false, err
}
if ref == nil {
s.bumpRedeemRetry(ctx, token, "上游未返回刷新结果", false)
return false, false, errors.New("上游未返回刷新结果")
}
now := time.Now()
upstreamStatus := strings.ToLower(strings.TrimSpace(ref.Token.Status))
// 上游明确失效:终止处理并标记异常。
if upstreamStatus == model.RockStatusAbnormal {
s.markRedeemingAbnormal(ctx, token, "上游状态异常(发货确认)", recorder)
return false, false, nil
}
if rockUpstreamRedeemed(ref) {
fulfillID := extractRockFulfillID(ref, rockClaimBaseline(token))
redeemedAt := now
if parsed, ok := parseRockTime(ref.Token.RedeemedAt); ok {
redeemedAt = parsed
}
fromStatus := token.Status
_ = s.db.WithContext(ctx).Unscoped().Model(token).Updates(map[string]interface{}{
"status": model.RockStatusRedeemed,
"upstream_fulfill_id": fulfillID,
"redeemed_at": redeemedAt,
"redeem_checked_at": now,
"redeem_next_retry_at": nil,
"redeem_last_message": "",
"redeem_last_error": "",
"redeem_timeout_warned_at": nil,
"updated_at": now,
}).Error
writeRockTokenStatusChange(s.db.WithContext(ctx), token, fromStatus, model.RockStatusRedeemed,
rockTokenSourceRedeem, "上游确认发货", false, nil)
token.Status = model.RockStatusRedeemed
token.UpstreamFulfillID = fulfillID
token.RedeemedAt = &redeemedAt
if recorder != nil {
recorder.Internal(ctx, "state_change", "success", "上游确认发货", map[string]interface{}{
"upstream_status": ref.Token.Status,
"upstream_redeemed_at": ref.Token.RedeemedAt,
"upstream_fulfill_id": fulfillID,
})
}
s.syncDeliveredOrder(ctx, token, recorder)
return true, false, nil
}
// 仍在官方处理中:保持 redeeming,指数退避。
waitingMessage := strings.TrimSpace(ref.Message)
if waitingMessage == "" {
waitingMessage = "官方处理中,请稍后刷新查看结果"
}
_ = 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
token.RedeemCheckedAt = &now
token.RedeemLastMessage = waitingMessage
attempts := token.RedeemAttempts
s.bumpRedeemRetry(ctx, token, "", true)
if recorder != nil {
recorder.Internal(ctx, "redeem_confirm", "pending", waitingMessage, map[string]interface{}{
"upstream_status": ref.Token.Status,
"attempts": attempts,
})
}
// 超过 15 分钟仍未确认:保持处理中,仅告警一次。
if token.RedeemRequestedAt != nil && now.Sub(*token.RedeemRequestedAt) >= rockRedeemTimeout && token.RedeemTimeoutWarnedAt == nil {
warnAt := now
_ = s.db.WithContext(ctx).Unscoped().Model(token).Updates(map[string]interface{}{
"redeem_timeout_warned_at": warnAt,
"redeem_last_message": waitingMessage + "(已超过 15 分钟未确认,已告警)",
}).Error
writeRockTokenStatusLog(s.db.WithContext(ctx), token, rockTokenEventStatusChanged,
model.RockStatusRedeeming, model.RockStatusRedeeming, rockTokenSourceRedeem,
"上游超过 15 分钟未确认发货,已触发告警", false, nil)
if recorder != nil {
recorder.Internal(ctx, "redeem_timeout", "pending", "上游超过 15 分钟未确认发货", nil)
}
return false, true, nil
}
return false, false, nil
}
func (s *RockService) refreshRedeemStatus(ctx context.Context, token *model.RockToken, bearerToken string) (*RockRefreshResponse, error) {
var firstErr error
if strings.TrimSpace(token.RedeemCode) != "" {
ref, err := s.client.RefreshRedeemBinding(ctx, token.RedeemCode)
if err == nil {
return ref, nil
}
firstErr = err
// 频控时不要再打管理端接口,直接交给退避。
if isRockRateLimitError(err) {
return nil, err
}
}
if strings.TrimSpace(token.UpstreamTokenID) != "" {
ref, err := s.client.RefreshToken(ctx, bearerToken, token.UpstreamTokenID)
if err == nil {
return ref, nil
}
if firstErr == nil {
firstErr = err
}
return nil, err
}
if firstErr != nil {
return nil, firstErr
}
return nil, errors.New("缺少兑换码与上游 Token ID,无法确认发货结果")
}
func (s *RockService) bumpRedeemRetry(ctx context.Context, token *model.RockToken, lastError string, countAttempt bool) {
if token == nil {
return
}
attempts := token.RedeemAttempts
if countAttempt {
attempts++
}
delay := rockRedeemFirstRetry
switch {
case attempts <= 1:
delay = rockRedeemFirstRetry
case attempts == 2:
delay = time.Minute
case attempts == 3:
delay = 2 * time.Minute
default:
delay = rockRedeemMaxRetry
}
if token.RedeemRequestedAt != nil && time.Since(*token.RedeemRequestedAt) >= rockRedeemTimeout {
delay = rockRedeemMaxRetry
}
next := time.Now().Add(delay)
updates := map[string]interface{}{
"redeem_attempts": attempts,
"redeem_next_retry_at": next,
"updated_at": time.Now(),
}
if strings.TrimSpace(lastError) != "" {
updates["redeem_last_error"] = lastError
}
_ = s.db.WithContext(ctx).Unscoped().Model(token).Updates(updates).Error
token.RedeemAttempts = attempts
token.RedeemNextRetryAt = &next
}
func (s *RockService) markRedeemingAbnormal(ctx context.Context, token *model.RockToken, reason string, recorder *RockDeliveryRecorder) {
fromStatus := token.Status
_ = 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
writeRockTokenStatusChange(s.db.WithContext(ctx), token, fromStatus, model.RockStatusAbnormal,
rockTokenSourceRedeem, reason, false, nil)
token.Status = model.RockStatusAbnormal
if recorder != nil {
recorder.Internal(ctx, "state_change", "failed", reason, map[string]interface{}{
"from_status": fromStatus,
"to_status": model.RockStatusAbnormal,
})
}
}
func (s *RockService) syncDeliveredOrder(ctx context.Context, token *model.RockToken, recorder *RockDeliveryRecorder) {
if s.redeemConfirmHook == nil || token == nil || strings.TrimSpace(token.AssignedOrderNo) == "" {
return
}
var order model.FulfillmentOrder
if err := s.db.WithContext(ctx).Where("order_no = ?", token.AssignedOrderNo).First(&order).Error; err != nil {
if recorder != nil {
recorder.Internal(ctx, "order_update", "failed", "读取订单失败: "+err.Error(), map[string]interface{}{
"order_no": token.AssignedOrderNo,
})
}
return
}
var lastErr error
for attempt := 0; attempt < 3; attempt++ {
if err := s.redeemConfirmHook(ctx, &order, token); err == nil {
if recorder != nil {
recorder.Internal(ctx, "order_update", "success", "订单已同步为已交付", map[string]interface{}{
"order_no": order.OrderNo,
})
}
return
} else {
lastErr = err
time.Sleep(time.Second)
}
}
if recorder != nil {
recorder.Internal(ctx, "order_update", "failed", fmt.Sprintf("同步订单失败: %v", lastErr), map[string]interface{}{
"order_no": order.OrderNo,
})
}
log.Printf("[Rock] Token %d 已确认发货,但订单 %s 同步失败: %v", token.ID, token.AssignedOrderNo, lastErr)
}
func appendRockRemark(existing, reason string) string {
existing = strings.TrimSpace(existing)
reason = strings.TrimSpace(reason)
if reason == "" {
return existing
}
if existing == "" {
return reason
}
if strings.Contains(existing, reason) {
return existing
}
return existing + "\n" + reason
}
@@ -0,0 +1,413 @@
package service
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"affiliate_dash/internal/model"
)
func setRockTestBearerToken(t *testing.T, svc *RockService, bearer string) {
t.Helper()
if _, err := svc.GetConfig(context.Background()); err != nil {
t.Fatalf("初始化配置失败: %v", err)
}
if err := svc.db.Model(&model.RockConfig{}).
Where("id = 1").
Update("bearer_token", bearer).Error; err != nil {
t.Fatalf("写入 Bearer Token 失败: %v", err)
}
}
// 上游返回 pending(HTTP 200 + status=pending + token.status=bound)时:
// 本地必须进入 redeeming,绝不能写 redeemed 终态或履约流水。
func TestRedeemTokenForOrder_PendingKeepsRedeeming(t *testing.T) {
db := newRockTestDB(t)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
switch {
case strings.HasSuffix(r.URL.Path, "/refresh"):
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"token": map[string]interface{}{
"id": 179384,
"status": "bound",
"bound_account": "舱z",
"bound_account_id": "3829515948",
"region": "安卓-微信",
"redeemed_at": nil,
},
"binding": map[string]interface{}{
"bound": true,
"role": "舱z",
"roleId": "3829515948",
"region": "安卓-微信",
},
"deliveryList": []interface{}{},
"message": "已刷新绑定信息",
})
case strings.HasPrefix(r.URL.Path, "/api/redeem/"):
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"token": map[string]interface{}{
"id": 179384,
"status": "bound",
"redeem_skin": nil,
"redeemed_at": nil,
"redeem_code": "CODE-PENDING",
"bound_account": "舱z",
},
"status": "pending",
"pending": true,
"message": "官方处理中,请稍后刷新查看结果",
})
default:
w.WriteHeader(http.StatusNotFound)
}
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearerToken(t, svc, "bearer-test")
now := time.Now()
token := model.RockToken{
UpstreamTokenID: "179384",
Token: "jwt-179384",
RedeemCode: "CODE-PENDING",
Color: model.RockColorBlack,
Status: model.RockStatusBound,
AssignedOrderNo: "FO-PENDING-1",
ExpectedUID: "3829515948",
ActualRoleName: "舱z",
ActualRoleID: "3829515948",
ActualChannel: "安卓-微信",
BoundAt: &now,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
recorder := svc.NewDeliveryRecorder(token.AssignedOrderNo, nil, "127.0.0.1", "go-test")
result, err := svc.RedeemTokenForOrder(context.Background(), token.AssignedOrderNo, model.RockSKUBlackTshirt, token.ExpectedUID, recorder)
if err != nil {
t.Fatalf("pending 不应返回错误: %v", err)
}
if result == nil || !result.Pending {
t.Fatalf("pending 响应应标记 Pending=true,实际: %+v", result)
}
var saved model.RockToken
if err := db.First(&saved, token.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if saved.Status != model.RockStatusRedeeming {
t.Fatalf("pending 后状态应为 redeeming,实际为 %s", saved.Status)
}
if saved.RedeemedAt != nil {
t.Fatalf("pending 不能写 redeemed_at,实际为 %v", saved.RedeemedAt)
}
if saved.UpstreamFulfillID != 0 {
t.Fatalf("pending 不能写履约流水,实际为 %d", saved.UpstreamFulfillID)
}
if saved.RedeemRequestedAt == nil || saved.RedeemNextRetryAt == nil {
t.Fatalf("pending 应记录提交时间与下次轮询时间: %+v", saved)
}
if strings.TrimSpace(saved.RedeemLastMessage) == "" {
t.Fatalf("pending 应保留上游处理中说明")
}
// 报文日志应完整记录兑换请求与响应(明文)。
var logs []model.RockDeliveryLog
if err := db.Where("token_id = ?", token.ID).Order("id ASC").Find(&logs).Error; err != nil {
t.Fatalf("读取发货日志失败: %v", err)
}
var hasRequest, hasPendingResponse bool
for _, entry := range logs {
if entry.Step == "redeem_request" && entry.Direction == "outbound" {
hasRequest = true
if entry.Status != "pending" {
t.Fatalf("兑换请求日志状态应为 pending,实际 %s", entry.Status)
}
if !strings.Contains(entry.RequestBody, "黑色ROCK长袖T恤") {
t.Fatalf("请求报文应包含皮肤名,实际 %q", entry.RequestBody)
}
if !strings.Contains(entry.ResponseBody, "官方处理中") {
t.Fatalf("响应报文应保留上游原文,实际 %q", entry.ResponseBody)
}
hasPendingResponse = true
}
}
if !hasRequest || !hasPendingResponse {
t.Fatalf("缺少兑换报文日志: %+v", logs)
}
}
// 上游 HTTP 200 但既非 success 也非 pending(例如空响应 / 状态异常)时必须报错且不改状态。
func TestRedeemTokenForOrder_UnexpectedPayloadDoesNotMarkRedeemed(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")
if strings.HasPrefix(r.URL.Path, "/api/redeem/") {
// 模拟上游业务失败但返回 200:只有 message,没有 status。
_, _ = w.Write([]byte(`{"token":{"status":"bound"},"message":"当前已禁止提交"}`))
return
}
w.WriteHeader(http.StatusNotFound)
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearerToken(t, svc, "bearer-test")
now := time.Now()
token := model.RockToken{
UpstreamTokenID: "179385",
Token: "jwt-179385",
RedeemCode: "CODE-BLOCKED",
Color: model.RockColorBlack,
Status: model.RockStatusBound,
AssignedOrderNo: "FO-BLOCKED-1",
ExpectedUID: "3829515949",
ActualRoleName: "舱y",
ActualRoleID: "3829515949",
BoundAt: &now,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
if _, err := svc.RedeemTokenForOrder(context.Background(), token.AssignedOrderNo, model.RockSKUBlackTshirt, token.ExpectedUID, nil); err == nil {
t.Fatalf("非 success/pending 响应必须报错,避免误判为已发货")
}
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("失败后状态应保持 bound,实际为 %s", saved.Status)
}
if saved.RedeemedAt != nil {
t.Fatalf("失败后不能写 redeemed_at")
}
}
// 已进入 redeeming 的 Token 再次提交必须直接返回处理中,不能重复调用上游兑换。
func TestRedeemTokenForOrder_RedeemingDoesNotResubmit(t *testing.T) {
db := newRockTestDB(t)
var redeemCalls int
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
if strings.HasPrefix(r.URL.Path, "/api/redeem/") {
redeemCalls++
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"status": "success",
"message": "领取成功",
"token": map[string]interface{}{"status": "redeemed"},
})
return
}
w.WriteHeader(http.StatusNotFound)
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearerToken(t, svc, "bearer-test")
now := time.Now()
next := now.Add(30 * time.Second)
token := model.RockToken{
UpstreamTokenID: "179386",
Token: "jwt-179386",
RedeemCode: "CODE-REDEEMING",
Color: model.RockColorBlack,
Status: model.RockStatusRedeeming,
AssignedOrderNo: "FO-REDEEMING-1",
ExpectedUID: "3829515950",
ActualRoleName: "舱x",
ActualRoleID: "3829515950",
BoundAt: &now,
RedeemRequestedAt: &now,
RedeemNextRetryAt: &next,
RedeemLastMessage: "官方处理中,请稍后刷新查看结果",
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
result, err := svc.RedeemTokenForOrder(context.Background(), token.AssignedOrderNo, model.RockSKUBlackTshirt, token.ExpectedUID, nil)
if err != nil {
t.Fatalf("重复提交应返回处理中而不是错误: %v", err)
}
if result == nil || !result.Pending {
t.Fatalf("重复提交应返回 Pending=true: %+v", result)
}
if redeemCalls != 0 {
t.Fatalf("redeeming 状态不应再次调用上游兑换,实际调用 %d 次", redeemCalls)
}
}
// 后台确认任务拿到 redeemed 后应写终态、记录履约流水并触发订单同步回调。
func TestProcessRedeemingTokens_ConfirmsAndSyncsOrder(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")
if strings.HasSuffix(r.URL.Path, "/refresh") {
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"token": map[string]interface{}{
"status": "redeemed",
"bound_account": "舱z",
"bound_account_id": "3829515948",
"region": "安卓-微信",
"redeemed_at": "2026-11-14 11:04:52",
},
"deliveryList": []interface{}{
map[string]interface{}{"fulfillId": 4519960, "time": "2026-11-14 11:04:52"},
},
"message": "已刷新绑定信息",
})
return
}
w.WriteHeader(http.StatusNotFound)
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearerToken(t, svc, "bearer-test")
var hookCalled int
var hookTokenID uint
var hookFulfillID int64
svc.SetRedeemConfirmHook(func(ctx context.Context, order *model.FulfillmentOrder, token *model.RockToken) error {
hookCalled++
hookTokenID = token.ID
hookFulfillID = token.UpstreamFulfillID
return nil
})
now := time.Now().Add(-2 * time.Minute)
next := now
token := model.RockToken{
UpstreamTokenID: "179384",
Token: "jwt-179384",
RedeemCode: "CODE-CONFIRM",
Color: model.RockColorBlack,
Status: model.RockStatusRedeeming,
AssignedOrderNo: "FO-CONFIRM-1",
ExpectedUID: "3829515948",
ActualRoleName: "舱z",
ActualRoleID: "3829515948",
BoundAt: &now,
RedeemRequestedAt: &now,
RedeemNextRetryAt: &next,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
order := model.FulfillmentOrder{
MerchantID: 1,
OrderNo: "FO-CONFIRM-1",
OrderStatus: model.OrderStatusDelivering,
}
if err := db.Create(&order).Error; err != nil {
t.Fatalf("创建订单失败: %v", err)
}
summary, err := svc.ProcessRedeemingTokens(context.Background())
if err != nil {
t.Fatalf("确认任务失败: %v", err)
}
if summary.Confirmed != 1 {
t.Fatalf("应确认 1 条,实际 %+v", summary)
}
var saved model.RockToken
if err := db.First(&saved, token.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if saved.Status != model.RockStatusRedeemed {
t.Fatalf("确认后状态应为 redeemed,实际 %s", saved.Status)
}
if saved.UpstreamFulfillID != 4519960 {
t.Fatalf("应记录履约流水 4519960,实际 %d", saved.UpstreamFulfillID)
}
if saved.RedeemedAt == nil {
t.Fatalf("确认后应写 redeemed_at")
}
if hookCalled != 1 || hookTokenID != token.ID || hookFulfillID != 4519960 {
t.Fatalf("订单同步回调不正确: called=%d token=%d fulfill=%d", hookCalled, hookTokenID, hookFulfillID)
}
}
// 上游仍为 bound 时应保持 redeeming,并按指数退避延后下一次查询。
func TestProcessRedeemingTokens_StaysPendingAndBacksOff(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")
if strings.HasSuffix(r.URL.Path, "/refresh") {
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"token": map[string]interface{}{
"status": "bound",
"bound_account": "舱z",
"bound_account_id": "3829515948",
"redeemed_at": nil,
},
"binding": map[string]interface{}{"bound": true, "roleId": "3829515948"},
"deliveryList": []interface{}{},
"message": "已刷新绑定信息",
})
return
}
w.WriteHeader(http.StatusNotFound)
}))
defer server.Close()
svc := NewRockService(db, NewRockClient(server.URL))
setRockTestBearerToken(t, svc, "bearer-test")
requested := time.Now().Add(-3 * time.Minute)
next := time.Now().Add(-time.Second)
token := model.RockToken{
UpstreamTokenID: "179387",
Token: "jwt-179387",
RedeemCode: "CODE-STILL-PENDING",
Color: model.RockColorBlack,
Status: model.RockStatusRedeeming,
AssignedOrderNo: "FO-STILL-PENDING",
ExpectedUID: "3829515948",
RedeemRequestedAt: &requested,
RedeemNextRetryAt: &next,
}
if err := db.Create(&token).Error; err != nil {
t.Fatalf("创建 Token 失败: %v", err)
}
summary, err := svc.ProcessRedeemingTokens(context.Background())
if err != nil {
t.Fatalf("确认任务失败: %v", err)
}
if summary.Confirmed != 0 || summary.Pending != 1 {
t.Fatalf("应保持处理中,实际 %+v", summary)
}
var saved model.RockToken
if err := db.First(&saved, token.ID).Error; err != nil {
t.Fatalf("读取 Token 失败: %v", err)
}
if saved.Status != model.RockStatusRedeeming {
t.Fatalf("状态应保持 redeeming,实际 %s", saved.Status)
}
if saved.RedeemAttempts != 1 {
t.Fatalf("轮询次数应为 1,实际 %d", saved.RedeemAttempts)
}
if saved.RedeemNextRetryAt == nil || !saved.RedeemNextRetryAt.After(time.Now()) {
t.Fatalf("应设置下一次退避时间: %+v", saved.RedeemNextRetryAt)
}
if saved.RedeemLastMessage == "" {
t.Fatalf("应保留上游处理中说明")
}
}
+20 -8
View File
@@ -23,6 +23,9 @@ type RockService struct {
refreshCache sync.Map // key: upstreamTokenID string, value: rockRefreshCacheItem
bindLinkCache sync.Map // key: token ID uint, value: rockBindLinkCacheItem
lastSummary *PurchaseSummary
// redeemConfirmHook 在上游确认发货后同步订单状态,由 main 装配 DeliveryService。
redeemConfirmHook func(ctx context.Context, order *model.FulfillmentOrder, token *model.RockToken) error
}
func NewRockService(db *gorm.DB, client *RockClient, codecs ...*SecretCodec) *RockService {
@@ -41,18 +44,20 @@ type RockStatsResponse struct {
AvailableStock int64 `json:"available_stock"`
UsingStock int64 `json:"using_stock"`
BoundStock int64 `json:"bound_stock"`
RedeemingStock int64 `json:"redeeming_stock"` // 上游官方处理中
RedeemedStock int64 `json:"redeemed_stock"`
MismatchStock int64 `json:"mismatch_stock"` // 绑定错误(待排查)
AbnormalStock int64 `json:"abnormal_stock"`
DeletedStock int64 `json:"deleted_stock"` // 回收站中已软删除的数量
// 过期预警计数:仅统计对应主状态(使用中 / 已绑定),用于筛选栏徽标。
UsingOverdueStock int64 `json:"using_overdue_stock"`
UsingDangerStock int64 `json:"using_danger_stock"`
BoundOverdueStock int64 `json:"bound_overdue_stock"`
UpstreamPoints float64 `json:"upstream_points"`
AutoDeliveryEnabled bool `json:"auto_delivery_enabled"`
TokenWarningStock int `json:"token_warning_stock"`
AutoPurchaseEnabled bool `json:"auto_purchase_enabled"`
// 过期预警计数:仅统计对应主状态(使用中 / 已绑定 / 发货处理中),用于筛选栏徽标。
UsingOverdueStock int64 `json:"using_overdue_stock"`
UsingDangerStock int64 `json:"using_danger_stock"`
BoundOverdueStock int64 `json:"bound_overdue_stock"`
RedeemingTimeoutStock int64 `json:"redeeming_timeout_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"`
// 自动补货冷却状态:供页面提示“下次可补货时间”
@@ -360,6 +365,8 @@ func (s *RockService) GetStats(ctx context.Context) (*RockStatsResponse, error)
stats.UsingStock = r.Total
case model.RockStatusBound:
stats.BoundStock = r.Total
case model.RockStatusRedeeming:
stats.RedeemingStock = r.Total
case model.RockStatusRedeemed:
stats.RedeemedStock = r.Total
case model.RockStatusMismatch:
@@ -396,6 +403,11 @@ func (s *RockService) GetStats(ctx context.Context) (*RockStatsResponse, error)
Count(&stats.BoundOverdueStock).Error; err != nil {
return nil, err
}
if err := s.db.WithContext(ctx).Model(&model.RockToken{}).
Where("status = ? AND redeem_requested_at IS NOT NULL AND redeem_requested_at <= ?", model.RockStatusRedeeming, now.Add(-rockRedeemTimeout)).
Count(&stats.RedeemingTimeoutStock).Error; err != nil {
return nil, err
}
if cfg.AutoPurchaseEnabled {
if remain := rockAutoPurchaseCooldownRemaining(cfg, time.Now()); remain > 0 {
@@ -23,9 +23,10 @@ const (
)
var rockTokenDangerousStatuses = map[string]bool{
model.RockStatusUsing: true,
model.RockStatusBound: true,
model.RockStatusMismatch: true,
model.RockStatusUsing: true,
model.RockStatusBound: true,
model.RockStatusRedeeming: true,
model.RockStatusMismatch: true,
}
// RockTokenDeleteStatusCount 是删除预览中的状态分布。
@@ -67,7 +68,7 @@ type RockTokenForceRequiredError struct {
}
func (e *RockTokenForceRequiredError) Error() string {
return fmt.Sprintf("本次操作包含 %d 条使用中/已绑定/绑定错误的 Token,必须二次确认后强制删除", e.Dangerous)
return fmt.Sprintf("本次操作包含 %d 条使用中/已绑定/发货处理中/绑定错误的 Token,必须二次确认后强制删除", e.Dangerous)
}
// RockTokenDeleteConflictError 表示预览后数据已变化。
@@ -84,7 +85,7 @@ func (e *RockTokenDeleteConflictError) Error() string {
func isKnownRockTokenStatus(status string) bool {
switch status {
case "", model.RockStatusAvailable, model.RockStatusUsing, model.RockStatusBound,
case "", model.RockStatusAvailable, model.RockStatusUsing, model.RockStatusBound, model.RockStatusRedeeming,
model.RockStatusRedeemed, model.RockStatusMismatch, model.RockStatusAbnormal,
rockTokenStatusUsingOverdue, rockTokenStatusBoundOverdue:
return true
@@ -186,6 +187,7 @@ func rockTokenStatusCountsFromRows(rows []model.RockToken) []RockTokenDeleteStat
model.RockStatusAvailable,
model.RockStatusUsing,
model.RockStatusBound,
model.RockStatusRedeeming,
model.RockStatusRedeemed,
model.RockStatusMismatch,
model.RockStatusAbnormal,