dachui/affiliate_dash
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(delivery): add transient retry, fix queue deadlock, and gate delivery submit by scan auth

Browse Source
This commit is contained in:
yml2213
2026-09-18 14:49:54 +08:00
parent 848d3da809
commit 20bde97d7a
8 changed files with 404 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/handler/delivery.go
+14
View File
@@ -53,6 +53,20 @@ func (h *DeliveryHandler) Bind(c *gin.Context) {
response.OK(c, result)
}
func (h *DeliveryHandler) GetBindResult(c *gin.Context) {
bindUUID := strings.TrimSpace(c.Query("bind_uuid"))
if bindUUID == "" {
response.BadRequest(c, "bind_uuid 不能为空")
return
}
result, err := h.deliverySvc.GetBindResultLinked(c.Param("order_no"), bindUUID, parseDeliveryAuth(c))
if err != nil {
writeDeliveryError(c, err)
return
}
response.OK(c, result)
}
type deliverySubmitReq struct {
GameAccount string `json:"game_account" binding:"required"`
BindUUID string `json:"bind_uuid" binding:"required"`
backend/internal/router/router.go
+1
View File
@@ -111,6 +111,7 @@ func Setup(h *Handlers) *gin.Engine {
{
delivery.GET("/orders/:order_no", h.Delivery.GetOrder)
delivery.POST("/orders/:order_no/bind", h.Delivery.Bind)
delivery.GET("/orders/:order_no/bind-result", h.Delivery.GetBindResult)
delivery.POST("/orders/:order_no/submit", h.Delivery.Submit)
delivery.GET("/image-proxy", h.Delivery.ImageProxy)
delivery.HEAD("/image-proxy", h.Delivery.ImageProxy)
backend/internal/service/delivery.go
+36
View File
@@ -383,6 +383,42 @@ func (s *DeliveryService) GetBindResult(merchantID uint, orderNo, bindUUID strin
}, nil
}
// GetBindResultLinked 为公开发货链接场景提供查询绑定结果的能力。
func (s *DeliveryService) GetBindResultLinked(orderNo, bindUUID string, auth DeliveryLinkAuth) (*DeliveryBindResult, error) {
bindUUID = strings.TrimSpace(bindUUID)
if bindUUID == "" {
return nil, errors.New("bind_uuid 不能为空")
}
openOrder, order, goodID, err := s.loadLinkedOrder(orderNo, &auth)
if err != nil {
return nil, err
}
if _, _, _, err := s.buildDeliveryState(openOrder, order, goodID, false); err != nil {
return nil, err
}
providers, _ := s.router.ResolveProviders(order)
if len(providers) == 0 {
return nil, errors.New("未配置发货渠道")
}
p := providers[0]
res, err := p.CheckBind(context.Background(), order, bindUUID)
if err != nil {
return nil, err
}
return &DeliveryBindResult{
BindUUID: res.BindUUID,
BindURL: res.BindURL,
QRURL: res.QRURL,
Bound: res.Bound,
GameAccount: res.GameAccount,
RoleName: res.RoleName,
GameChannel: res.GameChannel,
ExpectedGameAccount: res.ExpectedGameAccount,
Mismatch: res.Mismatch,
Message: res.Message,
}, nil
}
func (s *DeliveryService) bindWithOrder(orderNo, gameAccount string, auth *DeliveryLinkAuth, merchantID uint) (*DeliveryBindResult, error) {
gameAccount = strings.TrimSpace(gameAccount)
if gameAccount == "" {
backend/internal/service/provider_dlc.go
+123 -31
View File
@@ -215,12 +215,45 @@ func (p *DLCProvider) Submit(ctx context.Context, order *model.FulfillmentOrder,
}
if err := p.patchOrderQueue(queueOrderID, gameAccount, bindUUID, goodsNo, goodsName); err != nil {
return &ProviderSubmitResult{PatchResultData: partial}, fmt.Errorf("补充发货账号失败: %w", err)
if isQueueDeadError(err) {
// 旧队列已失效/不可操作，清除失效队列并重新创建
delete(partial, "queue_order_id")
newQueueID, createErr := p.createOrderQueue(goodID, order.OrderNo, goodsNo)
if createErr != nil {
return &ProviderSubmitResult{PatchResultData: partial}, fmt.Errorf("重置发货队列失败: %w (原错误: %v)", createErr, err)
}
queueOrderID = newQueueID
partial["queue_order_id"] = queueOrderID
if patchErr := p.patchOrderQueue(queueOrderID, gameAccount, bindUUID, goodsNo, goodsName); patchErr != nil {
return &ProviderSubmitResult{PatchResultData: partial}, fmt.Errorf("补充发货账号失败: %w", patchErr)
}
} else {
return &ProviderSubmitResult{PatchResultData: partial}, fmt.Errorf("补充发货账号失败: %w", err)
}
}
upstreamOrder, err := p.createUpstreamOrder(queueOrderID, gameAccount, goodID, order.OrderNo)
if err != nil {
return &ProviderSubmitResult{PatchResultData: partial}, fmt.Errorf("创建上游正式订单失败: %w", err)
if isQueueDeadError(err) {
// 创建正式单阶段出现订单不可操作，尝试重建队列重试一次
delete(partial, "queue_order_id")
newQueueID, createErr := p.createOrderQueue(goodID, order.OrderNo, goodsNo)
if createErr == nil {
queueOrderID = newQueueID
partial["queue_order_id"] = queueOrderID
if patchErr := p.patchOrderQueue(queueOrderID, gameAccount, bindUUID, goodsNo, goodsName); patchErr == nil {
if retriedOrder, retriedErr := p.createUpstreamOrder(queueOrderID, gameAccount, goodID, order.OrderNo); retriedErr == nil {
upstreamOrder = retriedOrder
err = nil
} else {
err = retriedErr
}
}
}
}
if err != nil {
return &ProviderSubmitResult{PatchResultData: partial}, fmt.Errorf("创建上游正式订单失败: %w", err)
}
}
providerOrderNo := firstNonEmpty(
@@ -366,6 +399,34 @@ func (p *DLCProvider) SyncStatus(ctx context.Context, order *model.FulfillmentOr
return nil, nil
}
func isQueueDeadError(err error) bool {
if err == nil {
return false
}
msg := err.Error()
return strings.Contains(msg, "订单不可操作") ||
strings.Contains(msg, "不可操作") ||
strings.Contains(msg, "订单不存在") ||
strings.Contains(msg, "队列已失效")
}
func isTransientDeliveryError(err error, statusCode int, message string) bool {
if err != nil {
return true
}
if statusCode == http.StatusTooManyRequests || statusCode == http.StatusBadGateway ||
statusCode == http.StatusServiceUnavailable || statusCode == http.StatusGatewayTimeout {
return true
}
msg := strings.ToLower(message)
return strings.Contains(msg, "请求过于频繁") ||
strings.Contains(msg, "稍后再试") ||
strings.Contains(msg, "系统繁忙") ||
strings.Contains(msg, "网络繁忙") ||
strings.Contains(msg, "too many requests") ||
strings.Contains(msg, "rate limit")
}
func (p *DLCProvider) signProxy(path, method string, data interface{}, out interface{}) error {
payload, err := json.Marshal(map[string]interface{}{
"path": path,
@@ -375,38 +436,69 @@ func (p *DLCProvider) signProxy(path, method string, data interface{}, out inter
if err != nil {
return err
}
req, err := http.NewRequest(http.MethodPost, p.bffBaseURL+"/sign-proxy", bytes.NewReader(payload))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
resp, err := p.httpClient.Do(req)
if err != nil {
return fmt.Errorf("发货服务请求失败：%w", err)
}
defer resp.Body.Close()
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
var envelope struct {
Code interface{} `json:"code"`
Message string `json:"message"`
Data json.RawMessage `json:"data"`
}
if err := json.Unmarshal(body, &envelope); err != nil {
return fmt.Errorf("发货服务响应无法解析：%s", truncateDeliveryText(string(body)))
}
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices || !isZeroCode(envelope.Code) {
if envelope.Message != "" {
return errors.New(envelope.Message)
const maxRetries = 2
var lastErr error
for attempt := 0; attempt <= maxRetries; attempt++ {
if attempt > 0 {
// 遭遇频控或瞬时网络波动时的指数避让 (1.2s, 2.4s)
backoff := time.Duration(1200*attempt)*time.Millisecond + time.Duration((attempt*137)%300)*time.Millisecond
time.Sleep(backoff)
}
req, err := http.NewRequest(http.MethodPost, p.bffBaseURL+"/sign-proxy", bytes.NewReader(payload))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
resp, err := p.httpClient.Do(req)
if err != nil {
lastErr = fmt.Errorf("发货服务请求失败：%w", err)
if attempt < maxRetries && isTransientDeliveryError(err, 0, "") {
continue
}
return lastErr
}
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
resp.Body.Close()
var envelope struct {
Code interface{} `json:"code"`
Message string `json:"message"`
Data json.RawMessage `json:"data"`
}
if err := json.Unmarshal(body, &envelope); err != nil {
lastErr = fmt.Errorf("发货服务响应无法解析：%s", truncateDeliveryText(string(body)))
if attempt < maxRetries && resp.StatusCode >= http.StatusInternalServerError {
continue
}
return lastErr
}
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices || !isZeroCode(envelope.Code) {
errMsg := envelope.Message
if errMsg == "" {
errMsg = fmt.Sprintf("HTTP %d", resp.StatusCode)
}
lastErr = errors.New(errMsg)
if attempt < maxRetries && isTransientDeliveryError(nil, resp.StatusCode, errMsg) {
continue
}
return lastErr
}
if out == nil || len(envelope.Data) == 0 || string(envelope.Data) == "null" {
return nil
}
if err := json.Unmarshal(envelope.Data, out); err != nil {
return fmt.Errorf("发货服务数据无法解析：%w", err)
}
return fmt.Errorf("发货服务请求失败：HTTP %d", resp.StatusCode)
}
if out == nil || len(envelope.Data) == 0 || string(envelope.Data) == "null" {
return nil
}
if err := json.Unmarshal(envelope.Data, out); err != nil {
return fmt.Errorf("发货服务数据无法解析：%w", err)
}
return nil
return lastErr
}
func timeSecond(s int) time.Duration {
backend/internal/service/provider_dlc_test.go
+115
View File
@@ -0,0 +1,115 @@
package service
import (
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
"time"
)
func TestIsQueueDeadError(t *testing.T) {
tests := []struct {
err error
expected bool
}{
{nil, false},
{errors.New("订单不可操作"), true},
{errors.New("补充发货账号失败: 订单不可操作"), true},
{errors.New("订单不存在"), true},
{errors.New("队列已失效"), true},
{errors.New("账号尚未绑定"), false},
{errors.New("请求过于频繁"), false},
}
for _, tc := range tests {
got := isQueueDeadError(tc.err)
if got != tc.expected {
t.Errorf("isQueueDeadError(%v) = %v, expected %v", tc.err, got, tc.expected)
}
}
}
func TestIsTransientDeliveryError(t *testing.T) {
tests := []struct {
err error
statusCode int
message string
expected bool
}{
{errors.New("dial tcp timeout"), 0, "", true},
{nil, 429, "", true},
{nil, 502, "", true},
{nil, 503, "", true},
{nil, 504, "", true},
{nil, 400, "绑定校验失败: 请求过于频繁，请稍后再试", true},
{nil, 400, "请求过于频繁", true},
{nil, 400, "系统繁忙，请稍后再试", true},
{nil, 400, "网络繁忙", true},
{nil, 400, "账号尚未绑定，请扫码完成绑定后再提交", false},
{nil, 400, "订单不可操作", false},
}
for _, tc := range tests {
got := isTransientDeliveryError(tc.err, tc.statusCode, tc.message)
if got != tc.expected {
t.Errorf("isTransientDeliveryError(%v, %d, %q) = %v, expected %v", tc.err, tc.statusCode, tc.message, got, tc.expected)
}
}
}
func TestSignProxyRetriesOnRateLimitAndSucceeds(t *testing.T) {
var callCount int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
count := atomic.AddInt32(&callCount, 1)
w.Header().Set("Content-Type", "application/json")
if count == 1 {
// 第一次调用返回频控限流
w.WriteHeader(http.StatusBadRequest)
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"code": 1,
"message": "请求过于频繁，请稍后再试",
})
return
}
// 第二次重试成功
w.WriteHeader(http.StatusOK)
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"code": 0,
"message": "success",
"data": map[string]interface{}{
"gameAccount": map[string]interface{}{
"game_account": "12345678",
},
},
})
}))
defer server.Close()
client := &http.Client{Timeout: 5 * time.Second}
p := NewDLCProvider(server.URL, "dlc", client)
var out struct {
GameAccount map[string]interface{} `json:"gameAccount"`
}
err := p.signProxy("/public/games/account-bound", "POST", map[string]interface{}{
"bind_uuid": "test-uuid",
}, &out)
if err != nil {
t.Fatalf("signProxy 预期在退避重试后成功，但返回错误: %v", err)
}
finalCount := atomic.LoadInt32(&callCount)
if finalCount != 2 {
t.Fatalf("预期总共调用 2 次（1次失败 + 1次重试成功），实际调用: %d", finalCount)
}
if out.GameAccount == nil || out.GameAccount["game_account"] != "12345678" {
t.Fatalf("预期返回账号信息，实际: %v", out.GameAccount)
}
}