yml/affiliate_dash
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化订单履约状态处理

Browse Source
This commit is contained in:
yml2213
2026-07-31 16:28:30 +08:00
parent a091806283
commit 677bdbbc1c
11 changed files with 488 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/service/fulfillment.go
+167 -36
View File
@@ -1,9 +1,11 @@
package service
import (
"context"
"encoding/json"
"errors"
"fmt"
"log"
"math"
"strings"
"time"
@@ -76,12 +78,16 @@ func (s *FulfillmentService) CreateOrder(in CreateFulfillmentOrderInput) (*Creat
}
requestData = string(raw)
}
fingerprint := orderRequestFingerprint(in, requestData)
result := &CreateFulfillmentOrderResult{}
err := s.db.Transaction(func(tx *gorm.DB) error {
var existing model.FulfillmentOrder
err := tx.Where("merchant_id = ? AND client_order_no = ?", in.MerchantID, in.ClientOrderNo).First(&existing).Error
if err == nil {
if err := ensureSameIdempotentOrder(&existing, fingerprint); err != nil {
return err
}
result.Order = &existing
result.Idempotent = true
return nil
@@ -139,24 +145,25 @@ func (s *FulfillmentService) CreateOrder(in CreateFulfillmentOrderInput) (*Creat
}
order := &model.FulfillmentOrder{
MerchantID: in.MerchantID,
OrderNo: newFulfillmentOrderNo(),
ClientOrderNo: in.ClientOrderNo,
MerchantProductID: product.ID,
ProductSKU: product.SKU,
ProductName: fallbackName(product.DisplayName, product.Product.Name),
Quantity: in.Quantity,
BaseAmount: baseAmount,
FeeType: merchant.FeeType,
FeeRateBP: merchant.FeeRateBP,
FeeFixedAmount: merchant.FeeFixedAmount,
ServiceFeeAmount: serviceFee,
Amount: totalAmount,
Currency: product.Currency,
PaymentStatus: model.PaymentStatusPaid,
FulfillmentStatus: model.FulfillmentStatusPending,
BuyerReference: in.BuyerReference,
RequestData: requestData,
MerchantID: in.MerchantID,
OrderNo: newFulfillmentOrderNo(),
ClientOrderNo: in.ClientOrderNo,
MerchantProductID: product.ID,
ProductSKU: product.SKU,
ProductName: fallbackName(product.DisplayName, product.Product.Name),
Quantity: in.Quantity,
BaseAmount: baseAmount,
FeeType: merchant.FeeType,
FeeRateBP: merchant.FeeRateBP,
FeeFixedAmount: merchant.FeeFixedAmount,
ServiceFeeAmount: serviceFee,
Amount: totalAmount,
Currency: product.Currency,
PaymentStatus: model.PaymentStatusPaid,
FulfillmentStatus: model.FulfillmentStatusPending,
BuyerReference: in.BuyerReference,
RequestFingerprint: fingerprint,
RequestData: requestData,
}
if err := tx.Create(order).Error; err != nil {
return err
@@ -198,6 +205,9 @@ func (s *FulfillmentService) CreateOrder(in CreateFulfillmentOrderInput) (*Creat
if strings.Contains(err.Error(), "duplicate") || strings.Contains(err.Error(), "UNIQUE") {
var existing model.FulfillmentOrder
if queryErr := s.db.Where("merchant_id = ? AND client_order_no = ?", in.MerchantID, in.ClientOrderNo).First(&existing).Error; queryErr == nil {
if sameErr := ensureSameIdempotentOrder(&existing, fingerprint); sameErr != nil {
return nil, sameErr
}
return &CreateFulfillmentOrderResult{Order: &existing, Idempotent: true}, nil
}
}
@@ -377,18 +387,12 @@ func (s *FulfillmentService) UpdateFulfillment(in FulfillmentUpdateInput) (*mode
}
return err
}
if order.FulfillmentStatus == model.FulfillmentStatusCancelled {
return errors.New("订单已取消，不能更新履约状态")
}
if order.PaymentStatus != model.PaymentStatusPaid {
return errors.New("订单未支付，不能履约")
}
if order.FulfillmentStatus == model.FulfillmentStatusSucceeded && in.Status == model.FulfillmentStatusSucceeded {
out = order
return nil
}
if order.FulfillmentStatus == model.FulfillmentStatusSucceeded {
return errors.New("订单已履约成功，不能回退状态")
if err := validateFulfillmentTransition(&order, in.Status, fulfillmentTransitionUpdate); err != nil {
return err
}
now := time.Now()
updates := map[string]interface{}{
@@ -427,6 +431,108 @@ func (s *FulfillmentService) UpdateFulfillment(in FulfillmentUpdateInput) (*mode
return &out, nil
}
func (s *FulfillmentService) MarkProcessingTimeouts(timeout time.Duration, limit int) (int, error) {
if timeout <= 0 {
return 0, nil
}
if limit <= 0 || limit > 100 {
limit = 50
}
now := time.Now()
cutoff := now.Add(-timeout)
var ids []uint
if err := s.db.Model(&model.FulfillmentOrder{}).
Where("payment_status = ? AND fulfillment_status = ? AND updated_at < ?", model.PaymentStatusPaid, model.FulfillmentStatusProcessing, cutoff).
Order("updated_at ASC, id ASC").
Limit(limit).
Pluck("id", &ids).Error; err != nil {
return 0, err
}
changed := 0
for _, id := range ids {
updated, err := s.markProcessingTimeout(id, timeout, now)
if err != nil {
return changed, err
}
if updated {
changed++
}
}
return changed, nil
}
func (s *FulfillmentService) markProcessingTimeout(id uint, timeout time.Duration, now time.Time) (bool, error) {
returned := false
err := s.db.Transaction(func(tx *gorm.DB) error {
var order model.FulfillmentOrder
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&order, id).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
return err
}
if !processingTimedOut(&order, timeout, now) {
return nil
}
if err := validateFulfillmentTransition(&order, model.FulfillmentStatusFailed, fulfillmentTransitionTimeout); err != nil {
return nil
}
reason := fmt.Sprintf("履约超时：订单已处于 processing 超过 %d 分钟", int(timeout.Minutes()))
updates := map[string]interface{}{
"fulfillment_status": model.FulfillmentStatusFailed,
"failure_reason": reason,
"result_data": buildProcessingTimeoutResultData(order.ResultData, timeout, now),
}
if err := tx.Model(&order).Updates(updates).Error; err != nil {
return err
}
var out model.FulfillmentOrder
if err := tx.First(&out, order.ID).Error; err != nil {
return err
}
if err := writeAudit(tx, &order.MerchantID, nil, nil, "fulfillment.timeout", "fulfillment_order", order.OrderNo, map[string]interface{}{
"from": order.FulfillmentStatus,
"to": model.FulfillmentStatusFailed,
"timeout_minutes": int(timeout.Minutes()),
"reason": reason,
}); err != nil {
return err
}
if s.callbacks != nil {
if err := s.callbacks.Enqueue(tx, order.MerchantID, "order.fulfillment.updated", orderCallbackData(&out)); err != nil {
return err
}
}
returned = true
return nil
})
return returned, err
}
func (s *FulfillmentService) RunProcessingTimeoutMonitor(ctx context.Context, timeout, interval time.Duration) {
if timeout <= 0 {
return
}
if interval <= 0 {
interval = time.Minute
}
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
changed, err := s.MarkProcessingTimeouts(timeout, 50)
if err != nil {
log.Printf("[fulfillment] timeout scan error: %v", err)
} else if changed > 0 {
log.Printf("[fulfillment] timeout scan marked failed count=%d", changed)
}
select {
case <-ctx.Done():
return
case <-ticker.C:
}
}
}
func (s *FulfillmentService) CancelOrder(merchantID, apiClientID uint, orderNo, reason string) (*model.FulfillmentOrder, error) {
var out model.FulfillmentOrder
err := s.db.Transaction(func(tx *gorm.DB) error {
@@ -443,8 +549,8 @@ func (s *FulfillmentService) CancelOrder(merchantID, apiClientID uint, orderNo,
out = order
return nil
}
if order.FulfillmentStatus == model.FulfillmentStatusProcessing || order.FulfillmentStatus == model.FulfillmentStatusSucceeded {
return errors.New("订单已进入履约流程，不能取消")
if err := canCancelOrder(&order); err != nil {
return err
}
now := time.Now()
updates := map[string]interface{}{
@@ -1030,13 +1136,15 @@ func (s *FulfillmentService) HandleShipNotify(in ShipNotifyInput) (*ShipNotifyRe
var message string
switch in.ShipStatus {
case "success":
if order.FulfillmentStatus != model.FulfillmentStatusPending &&
order.FulfillmentStatus != model.FulfillmentStatusFailed &&
order.FulfillmentStatus != model.FulfillmentStatusProcessing {
if err := writeShipNotifyRejectedAudit(tx, &order, in, "当前状态不允许标记发货成功"); err != nil {
if err := validateFulfillmentTransition(&order, nextStatus, fulfillmentTransitionShipNotify); err != nil {
message := "当前状态不允许标记发货成功"
if order.PaymentStatus != model.PaymentStatusPaid {
message = "订单未支付，拒绝成功推送"
}
if err := writeShipNotifyRejectedAudit(tx, &order, in, message); err != nil {
return err
}
rejectionErr = fmt.Errorf("当前状态 %s 不允许标记发货成功", order.FulfillmentStatus)
rejectionErr = err
return nil
}
updates["delivered_at"] = shippedAt
@@ -1046,11 +1154,17 @@ func (s *FulfillmentService) HandleShipNotify(in ShipNotifyInput) (*ShipNotifyRe
}
message = "发货成功，订单已交付"
case "failed":
if order.FulfillmentStatus == model.FulfillmentStatusSucceeded {
if err := writeShipNotifyRejectedAudit(tx, &order, in, "订单已交付，拒绝失败推送"); err != nil {
if err := validateFulfillmentTransition(&order, nextStatus, fulfillmentTransitionShipNotify); err != nil {
message := "当前状态不允许标记发货失败"
if order.FulfillmentStatus == model.FulfillmentStatusSucceeded {
message = "订单已交付，拒绝失败推送"
} else if order.PaymentStatus != model.PaymentStatusPaid {
message = "订单未支付，拒绝失败推送"
}
if err := writeShipNotifyRejectedAudit(tx, &order, in, message); err != nil {
return err
}
rejectionErr = errors.New("订单已交付，不能标记发货失败")
rejectionErr = err
return nil
}
updates["failure_reason"] = in.FailReason
@@ -1150,6 +1264,23 @@ func buildShipNotifyResultData(existing string, in ShipNotifyInput, shippedAt *t
return string(raw)
}
func buildProcessingTimeoutResultData(existing string, timeout time.Duration, now time.Time) string {
m := map[string]interface{}{}
if existing != "" && json.Valid([]byte(existing)) {
_ = json.Unmarshal([]byte(existing), &m)
}
m["timeout"] = true
m["timeout_minutes"] = int(timeout.Minutes())
m["timeout_at"] = timeutil.FormatAPITime(now)
m["ship_status"] = "failed"
m["fail_reason"] = fmt.Sprintf("履约超时：订单已处于 processing 超过 %d 分钟", int(timeout.Minutes()))
raw, err := json.Marshal(m)
if err != nil {
return existing
}
return string(raw)
}
// extractGameFields 从 JSON 文本中还原游戏相关字段（仅填充当前为空的字段）。
func extractGameFields(raw string, out *OpenOrderQuery) {
if raw == "" || !json.Valid([]byte(raw)) {