187 lines
6.6 KiB
Go
187 lines
6.6 KiB
Go
package purchase
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/google/uuid"
|
||
"go-admin/app/goauto/models"
|
||
"gorm.io/gorm"
|
||
"gorm.io/gorm/clause"
|
||
"gorm.io/gorm/logger"
|
||
)
|
||
|
||
type OrderWritebackView struct {
|
||
Status string `json:"status"`
|
||
Reason string `json:"reason,omitempty"`
|
||
CompletedAt *time.Time `json:"completedAt,omitempty"`
|
||
CanSubmit bool `json:"canSubmit"`
|
||
}
|
||
type OrderWritebackRequest struct {
|
||
RequestID string `json:"requestId"`
|
||
PurchaseTaskIDs []uint64 `json:"purchaseTaskIds"`
|
||
}
|
||
type OrderWritebackAcceptance struct {
|
||
TaskID uint64 `json:"taskId"`
|
||
Result string `json:"result"`
|
||
Reason string `json:"reason,omitempty"`
|
||
}
|
||
type OrderWritebackResponse struct {
|
||
Items []OrderWritebackAcceptance `json:"items"`
|
||
}
|
||
|
||
func orderWritebackEligible(t models.PurchaseTask) bool {
|
||
return t.ExecutionMode == models.PurchaseExecutionModeLive && t.TaskType == models.PurchaseTaskTypeSYBOrder && t.Status == models.PurchaseTaskStatusOrderCreated && t.SYBProductID != nil && t.PDDOrderNo != nil && strings.TrimSpace(*t.PDDOrderNo) != ""
|
||
}
|
||
|
||
// Called within the same transaction as the order fact. Never contacts SYB.
|
||
func ensureOrderWriteback(tx *gorm.DB, t models.PurchaseTask) error {
|
||
if !orderWritebackEligible(t) {
|
||
return nil
|
||
}
|
||
var syb models.SYBProduct
|
||
if err := tx.First(&syb, *t.SYBProductID).Error; err != nil {
|
||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return nil
|
||
}
|
||
return err
|
||
}
|
||
if syb.StockID <= 0 || syb.DetailID <= 0 {
|
||
return nil
|
||
}
|
||
if syb.StockID > 1<<63-1 || syb.DetailID > 1<<63-1 {
|
||
return nil
|
||
}
|
||
row := models.PurchaseOrderWriteback{PurchaseTaskID: t.ID, StockID: int64(syb.StockID), DetailID: int64(syb.DetailID), OrderNo: *t.PDDOrderNo, Status: "pending"}
|
||
return tx.Session(&gorm.Session{Logger: logger.Default.LogMode(logger.Silent)}).Clauses(clause.OnConflict{DoNothing: true}).Create(&row).Error
|
||
}
|
||
|
||
func (s *Service) OrderWritebackViews(ctx context.Context, tasks []models.PurchaseTask) (map[uint64]OrderWritebackView, error) {
|
||
out := map[uint64]OrderWritebackView{}
|
||
ids := make([]uint64, 0, len(tasks))
|
||
for _, t := range tasks {
|
||
ids = append(ids, t.ID)
|
||
v := OrderWritebackView{Status: "not_applicable", Reason: "不符合正式SYB订单回填条件"}
|
||
if orderWritebackEligible(t) {
|
||
v = OrderWritebackView{Status: "not_started", CanSubmit: true}
|
||
}
|
||
out[t.ID] = v
|
||
}
|
||
if len(ids) == 0 {
|
||
return out, nil
|
||
}
|
||
var rows []models.PurchaseOrderWriteback
|
||
if err := s.DB.WithContext(ctx).Where("purchase_task_id IN ?", ids).Find(&rows).Error; err != nil {
|
||
return nil, internal(err)
|
||
}
|
||
for _, r := range rows {
|
||
v := out[r.PurchaseTaskID]
|
||
v.Status = r.Status
|
||
v.Reason = r.ErrorMessage
|
||
v.CompletedAt = r.CompletedAt
|
||
v.CanSubmit = v.CanSubmit && (r.Status == "failed" || r.Status == "unknown") && (r.LeaseExpiresAt == nil || !r.LeaseExpiresAt.After(s.Now()))
|
||
out[r.PurchaseTaskID] = v
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (s *Service) RequestOrderWriteback(ctx context.Context, req OrderWritebackRequest) (OrderWritebackResponse, error) {
|
||
out := OrderWritebackResponse{Items: []OrderWritebackAcceptance{}}
|
||
if uuid.Validate(req.RequestID) != nil || len(req.PurchaseTaskIDs) == 0 || len(req.PurchaseTaskIDs) > 100 {
|
||
return out, fail(CodeInvalidRequest, "请选择1～100条采购任务")
|
||
}
|
||
ids := append([]uint64(nil), req.PurchaseTaskIDs...)
|
||
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
|
||
for i, id := range ids {
|
||
if id == 0 || (i > 0 && id == ids[i-1]) {
|
||
return out, fail(CodeInvalidRequest, "任务编号无效或重复")
|
||
}
|
||
}
|
||
data, _ := json.Marshal(ids)
|
||
sum := sha256.Sum256(data)
|
||
fingerprint := hex.EncodeToString(sum[:])
|
||
db := s.DB.WithContext(ctx).Session(&gorm.Session{Logger: logger.Default.LogMode(logger.Silent)})
|
||
err := db.Transaction(func(tx *gorm.DB) error {
|
||
// Serializes batch command replay without holding a lock across remote IO.
|
||
if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&models.PurchaseOrderWritebackLease{ID: 1}).Error; err != nil {
|
||
return err
|
||
}
|
||
var guard models.PurchaseOrderWritebackLease
|
||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&guard, 1).Error; err != nil {
|
||
return err
|
||
}
|
||
var command models.PurchaseOrderWritebackCommand
|
||
err := tx.First(&command, "request_id = ?", req.RequestID).Error
|
||
if err == nil {
|
||
if command.InputHash != fingerprint {
|
||
return fail(CodeInvalidRequest, "同一requestId不能改变任务集合")
|
||
}
|
||
return json.Unmarshal([]byte(command.ResultJSON), &out)
|
||
}
|
||
if !errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return err
|
||
}
|
||
for _, id := range ids {
|
||
a := OrderWritebackAcceptance{TaskID: id, Result: "skipped", Reason: "任务不存在或不符合回填条件"}
|
||
var task models.PurchaseTask
|
||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&task, id).Error; err != nil {
|
||
if !errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return err
|
||
}
|
||
out.Items = append(out.Items, a)
|
||
continue
|
||
}
|
||
if !orderWritebackEligible(task) {
|
||
out.Items = append(out.Items, a)
|
||
continue
|
||
}
|
||
if err := ensureOrderWriteback(tx, task); err != nil {
|
||
return err
|
||
}
|
||
var row models.PurchaseOrderWriteback
|
||
if err := tx.Where("purchase_task_id = ?", id).First(&row).Error; err != nil {
|
||
if !errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return err
|
||
}
|
||
a.Reason = "SYB明细关联已失效"
|
||
out.Items = append(out.Items, a)
|
||
continue
|
||
}
|
||
switch {
|
||
case row.OrderNo != *task.PDDOrderNo:
|
||
a.Result, a.Reason = "conflict", "订单号快照不一致，请人工核对"
|
||
case row.Status == "succeeded":
|
||
a.Result, a.Reason = "succeeded", "已回填"
|
||
case row.Status == "conflict":
|
||
a.Result, a.Reason = "conflict", row.ErrorMessage
|
||
case row.Status == "running" || (row.LeaseExpiresAt != nil && row.LeaseExpiresAt.After(s.Now())):
|
||
a.Reason = "正在回填或等待在途请求结束"
|
||
case row.Status == "pending":
|
||
a.Result, a.Reason = "pending", "已加入回填"
|
||
default:
|
||
if err := tx.Model(&row).Updates(map[string]any{"status": "pending", "write_started": false, "lease_owner": "", "lease_expires_at": nil, "error_code": "", "error_message": ""}).Error; err != nil {
|
||
return err
|
||
}
|
||
a.Result, a.Reason = "pending", "已加入回填，将先回读SYB"
|
||
}
|
||
out.Items = append(out.Items, a)
|
||
}
|
||
result, _ := json.Marshal(out)
|
||
return tx.Create(&models.PurchaseOrderWritebackCommand{RequestID: req.RequestID, InputHash: fingerprint, ResultJSON: string(result)}).Error
|
||
})
|
||
if err != nil {
|
||
var e *ServiceError
|
||
if errors.As(err, &e) {
|
||
return out, err
|
||
}
|
||
return out, internal(err)
|
||
}
|
||
return out, nil
|
||
}
|