diff --git a/backend/internal/pkg/response/response.go b/backend/internal/pkg/response/response.go index 2e240d9..506c07c 100644 --- a/backend/internal/pkg/response/response.go +++ b/backend/internal/pkg/response/response.go @@ -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}) } diff --git a/backend/internal/service/callback.go b/backend/internal/service/callback.go index 8b8659c..06faf89 100644 --- a/backend/internal/service/callback.go +++ b/backend/internal/service/callback.go @@ -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() } diff --git a/backend/internal/service/fulfillment.go b/backend/internal/service/fulfillment.go index 22e4a76..1a96b25 100644 --- a/backend/internal/service/fulfillment.go +++ b/backend/internal/service/fulfillment.go @@ -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 } diff --git a/backend/internal/service/provider_dlc.go b/backend/internal/service/provider_dlc.go index 9141320..3b26132 100644 --- a/backend/internal/service/provider_dlc.go +++ b/backend/internal/service/provider_dlc.go @@ -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++ { diff --git a/backend/internal/service/provider_dlc_test.go b/backend/internal/service/provider_dlc_test.go index 27824fb..d23d702 100644 --- a/backend/internal/service/provider_dlc_test.go +++ b/backend/internal/service/provider_dlc_test.go @@ -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