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

fix(delivery): 增加绑定缓存防频控、发货队列防重试死锁、终止无效回调重试

Browse Source
This commit is contained in:
yml2213
2026-09-18 15:26:09 +08:00
parent ec76844da8
commit 40194fa7e5
5 changed files with 135 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/pkg/response/response.go
+4
View File
@@ -2,6 +2,7 @@ package response
import (
"encoding/json"
"errors"
"fmt"
"net/http"
"reflect"
@@ -33,6 +34,9 @@ func WithStatus(c *gin.Context, httpStatus int, data interface{}) {
}
func Fail(c *gin.Context, httpStatus int, code int, message string) {
if message != "" && c != nil {
_ = c.Error(errors.New(message))
}
c.JSON(httpStatus, Body{Code: code, Message: message})
}
backend/internal/service/callback.go
+16 -2
View File
@@ -340,15 +340,29 @@ func (s *CallbackService) dispatchOne(ctx context.Context, id uint) (bool, bool,
"delivered_at": time.Now(),
}).Error
}
applog.Warnf("[callback] push fail event=%s event_id=%s merchant_id=%d url=%s status=%d body=%s", delivery.Event, delivery.EventID, delivery.MerchantID, delivery.CallbackSubscription.URL, resp.StatusCode, truncateCallbackError(string(body)))
if isTerminalCallbackError(resp.StatusCode, string(body)) {
applog.Infof("[callback] push terminal reject (no-retry) event=%s event_id=%s merchant_id=%d url=%s status=%d body=%s", delivery.Event, delivery.EventID, delivery.MerchantID, delivery.CallbackSubscription.URL, resp.StatusCode, truncateCallbackError(string(body)))
} else {
applog.Warnf("[callback] push fail event=%s event_id=%s merchant_id=%d url=%s status=%d body=%s", delivery.Event, delivery.EventID, delivery.MerchantID, delivery.CallbackSubscription.URL, resp.StatusCode, truncateCallbackError(string(body)))
}
return false, true, s.recordCallbackFailure(delivery, resp.StatusCode, string(body))
}
func isTerminalCallbackError(statusCode int, response string) bool {
if statusCode == http.StatusNotFound {
return true
}
respLower := strings.ToLower(response)
return strings.Contains(respLower, "affiliate_dash_callback_task_not_found") ||
strings.Contains(respLower, "未匹配到本地任务") ||
strings.Contains(respLower, "task_not_found")
}
func (s *CallbackService) recordCallbackFailure(delivery model.CallbackDelivery, statusCode int, response string) error {
attempts := delivery.Attempts + 1
status := model.CallbackDeliveryPending
nextAttempt := time.Now().Add(s.callbackRetryDelay(attempts))
if attempts >= s.maxAttempts {
if attempts >= s.maxAttempts || isTerminalCallbackError(statusCode, response) {
status = model.CallbackDeliveryFailed
nextAttempt = time.Now()
}
backend/internal/service/fulfillment.go
+1 -1
View File
@@ -244,7 +244,7 @@ func (s *FulfillmentService) CreateOrder(in CreateFulfillmentOrderInput) (*Creat
if err := writeAudit(tx, &in.MerchantID, optionalUint(in.ActorUserID), optionalUint(in.APIClientID), action, "fulfillment_order", order.OrderNo, map[string]interface{}{"client_order_no": in.ClientOrderNo, "sku": in.SKU, "source": orderSource}); err != nil {
return err
}
if s.callbacks != nil {
if s.callbacks != nil && orderSource != model.OrderSourceManual {
if err := s.callbacks.Enqueue(tx, in.MerchantID, "order.created", orderCallbackData(order)); err != nil {
return err
}
backend/internal/service/provider_dlc.go
+90 -15
View File
@@ -9,16 +9,25 @@ import (
"io"
"net/http"
"strings"
"sync"
"time"
"affiliate_dash/internal/model"
)
type bindCacheEntry struct {
account map[string]interface{}
err error
expiresAt time.Time
}
// DLCProvider 实现第一平台（聚鑫源 / dlc）履约逻辑
type DLCProvider struct {
bffBaseURL string
channel string
httpClient *http.Client
bffBaseURL string
channel string
httpClient *http.Client
bindCacheMu sync.RWMutex
bindCache map[string]bindCacheEntry
}
func NewDLCProvider(bffBaseURL, channel string, httpClient *http.Client) *DLCProvider {
@@ -37,6 +46,7 @@ func NewDLCProvider(bffBaseURL, channel string, httpClient *http.Client) *DLCPro
bffBaseURL: bffBaseURL,
channel: channel,
httpClient: httpClient,
bindCache: make(map[string]bindCacheEntry),
}
}
@@ -147,6 +157,14 @@ func (p *DLCProvider) CheckBind(ctx context.Context, order *model.FulfillmentOrd
Provider: ProviderDLC,
}, nil
}
if isRateLimitError(err) {
return &ProviderBindResult{
BindUUID: bindUUID,
Bound: false,
Message: "正在同步绑定状态，请稍候...",
Provider: ProviderDLC,
}, nil
}
return nil, err
}
@@ -312,25 +330,54 @@ func (p *DLCProvider) goodsDetail(goodID string) (map[string]interface{}, error)
}
func (p *DLCProvider) accountBound(bindUUID, goodID string) (map[string]interface{}, error) {
cacheKey := bindUUID + ":" + goodID
p.bindCacheMu.RLock()
if entry, ok := p.bindCache[cacheKey]; ok && time.Now().Before(entry.expiresAt) {
p.bindCacheMu.RUnlock()
return entry.account, entry.err
}
p.bindCacheMu.RUnlock()
var out struct {
GameAccount map[string]interface{} `json:"gameAccount"`
Snake map[string]interface{} `json:"game_account"`
}
if err := p.signProxy("/public/games/account-bound", "POST", map[string]interface{}{
err := p.signProxy("/public/games/account-bound", "POST", map[string]interface{}{
"bind_uuid": bindUUID,
"bindUuid": bindUUID,
"goodId": goodID,
"good_id": goodID,
}, &out); err != nil {
return nil, err
}, &out)
var account map[string]interface{}
ttl := 2 * time.Second
if err != nil {
if isRateLimitError(err) {
ttl = 3 * time.Second
}
} else {
account = out.GameAccount
if account == nil {
account = out.Snake
}
if account == nil {
err = errors.New("账号尚未绑定，请扫码完成绑定后再提交")
ttl = 2 * time.Second
} else {
ttl = 15 * time.Second
}
}
if out.GameAccount == nil {
out.GameAccount = out.Snake
p.bindCacheMu.Lock()
p.bindCache[cacheKey] = bindCacheEntry{
account: account,
err: err,
expiresAt: time.Now().Add(ttl),
}
if out.GameAccount == nil {
return nil, errors.New("账号尚未绑定，请扫码完成绑定后再提交")
}
return out.GameAccount, nil
p.bindCacheMu.Unlock()
return account, err
}
func (p *DLCProvider) createOrderQueue(goodID, orderNo, goodsNo string) (string, error) {
@@ -346,7 +393,19 @@ func (p *DLCProvider) createOrderQueue(goodID, orderNo, goodsNo string) (string,
if goodsNo != "" {
payload["goods_no"] = goodsNo
}
if err := p.signProxy("/public/users/orders-queue", "POST", payload, &out); err != nil {
err := p.signProxy("/public/users/orders-queue", "POST", payload, &out)
if err != nil {
if strings.Contains(err.Error(), "正在处理中") || strings.Contains(err.Error(), "请勿重复提交") {
// DLC 已存在此 order_sn 的队列，追加随机后缀重新创建队列
retrySN := fmt.Sprintf("%s_%x", orderNo, time.Now().UnixNano()%0xffff)
payload["order_sn"] = retrySN
out.Orders = nil
if retryErr := p.signProxy("/public/users/orders-queue", "POST", payload, &out); retryErr == nil && len(out.Orders) > 0 {
err = nil
}
}
}
if err != nil {
return "", err
}
if len(out.Orders) == 0 {
@@ -407,7 +466,20 @@ func isQueueDeadError(err error) bool {
return strings.Contains(msg, "订单不可操作") ||
strings.Contains(msg, "不可操作") ||
strings.Contains(msg, "订单不存在") ||
strings.Contains(msg, "队列已失效")
strings.Contains(msg, "队列已失效") ||
strings.Contains(msg, "正在处理中") ||
strings.Contains(msg, "请勿重复提交")
}
func isRateLimitError(err error) bool {
if err == nil {
return false
}
msg := strings.ToLower(err.Error())
return strings.Contains(msg, "请求过于频繁") ||
strings.Contains(msg, "稍后再试") ||
strings.Contains(msg, "too many requests") ||
strings.Contains(msg, "rate limit")
}
func isTransientDeliveryError(err error, statusCode int, message string) bool {
@@ -437,7 +509,10 @@ func (p *DLCProvider) signProxy(path, method string, data interface{}, out inter
return err
}
const maxRetries = 2
maxRetries := 2
if path == "/public/games/account-bound" {
maxRetries = 0 // 轮询绑定接口不触发重试睡眠，避免堆积阻塞
}
var lastErr error
for attempt := 0; attempt <= maxRetries; attempt++ {
backend/internal/service/provider_dlc_test.go
+24
View File
@@ -20,6 +20,8 @@ func TestIsQueueDeadError(t *testing.T) {
{errors.New("补充发货账号失败: 订单不可操作"), true},
{errors.New("订单不存在"), true},
{errors.New("队列已失效"), true},
{errors.New("该订单正在处理中，请勿重复提交"), true},
{errors.New("创建发货队列失败: 该订单正在处理中，请勿重复提交"), true},
{errors.New("账号尚未绑定"), false},
{errors.New("请求过于频繁"), false},
}
@@ -32,6 +34,28 @@ func TestIsQueueDeadError(t *testing.T) {
}
}
func TestIsRateLimitError(t *testing.T) {
tests := []struct {
err error
expected bool
}{
{nil, false},
{errors.New("请求过于频繁，请稍后再试"), true},
{errors.New("系统繁忙，请稍后再试"), true},
{errors.New("Rate limit exceeded"), true},
{errors.New("Too Many Requests"), true},
{errors.New("账号尚未绑定"), false},
{errors.New("订单不存在"), false},
}
for _, tc := range tests {
got := isRateLimitError(tc.err)
if got != tc.expected {
t.Errorf("isRateLimitError(%v) = %v, expected %v", tc.err, got, tc.expected)
}
}
}
func TestIsTransientDeliveryError(t *testing.T) {
tests := []struct {
err error