OPC/goauto
2
0
Fork 0
Code Issues 193 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(#147): move purchase matching outside transaction

Browse Source
This commit is contained in:
QiuSW
2026-08-29 11:09:48 +08:00
parent b453634b11
commit c9e9539de0
4 changed files with 304 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/app/goauto/purchase/batch.go
+32 -2
View File
@@ -209,7 +209,7 @@ func (s *Service) BatchCreate(ctx context.Context, req BatchCreateRequest) (Batc
var replay models.PurchaseTask
if err := s.DB.WithContext(ctx).Where("create_request_id = ?", itemRequestID).First(&replay).Error; err == nil {
taskID := replay.ID
preview := s.previewOne(ctx, id)
preview := replayBatchPreview(replay)
preview.Eligible, preview.ReasonCode, preview.Reason, preview.NextAction, preview.ActiveTaskID = true, "", "", "", nil
result.Items = append(result.Items, BatchCreateItem{BatchPreviewItem: preview, Created: true, TaskID: &taskID, TaskNo: taskNumber(taskID), Replayed: true})
result.CreatedCount++
@@ -218,7 +218,7 @@ func (s *Service) BatchCreate(ctx context.Context, req BatchCreateRequest) (Batc
return BatchCreateResponse{}, internal(err)
}
preview := s.previewOne(ctx, id)
preview := s.previewOneDeterministic(ctx, id)
if preview.Eligible && deviceErr != nil {
preview.Eligible = false
preview.ReasonCode, preview.Reason, preview.NextAction = serviceErrorFields(deviceErr)
@@ -294,6 +294,36 @@ func (s *Service) previewOne(ctx context.Context, id uint64) BatchPreviewItem {
return s.previewFromDataset(ctx, id, dataset, true)
}
func (s *Service) previewOneDeterministic(ctx context.Context, id uint64) BatchPreviewItem {
dataset, err := s.loadBatchPreviewDataset(ctx, []uint64{id})
if err != nil {
item := BatchPreviewItem{SYBProductID: id}
item.ReasonCode, item.Reason, item.NextAction = CodeInternal, "服务端处理失败", "retry"
return item
}
return s.previewFromDataset(ctx, id, dataset, false)
}
func replayBatchPreview(task models.PurchaseTask) BatchPreviewItem {
return BatchPreviewItem{
SYBProductID: taskPointerValue(task.SYBProductID), ShopeeProductID: task.ShopeeProductID,
PDDProductID: &task.PDDProductID, ProductTitle: task.ShopeeTitleSnapshot,
PDDGoodsID: task.PDDGoodsIDSnapshot, PDDTitle: task.PDDTitleSnapshot,
TargetColor: task.TargetColorSnapshot, TargetSize: task.TargetSizeSnapshot,
MappedColor: task.MappedColorSnapshot, MappedSize: task.MappedSizeSnapshot,
Quantity: task.Quantity, ReferenceUnitPriceCent: task.ReferenceUnitPriceCent,
MinUnitPriceCent: task.MinUnitPriceCent, MaxUnitPriceCent: task.MaxUnitPriceCent,
Currency: task.Currency, Eligible: true,
}
}
func taskPointerValue(value *uint64) uint64 {
if value == nil {
return 0
}
return *value
}
func (s *Service) previewFromDataset(ctx context.Context, id uint64, dataset batchPreviewDataset, allowAI bool) BatchPreviewItem {
item := BatchPreviewItem{SYBProductID: id}
syb, found := dataset.sybByID[id]
server/app/goauto/purchase/service.go
+165 -9
View File
@@ -7,6 +7,7 @@ import (
"encoding/json"
"errors"
"strings"
"sync"
"time"
"go-admin/app/goauto/aimatching"
@@ -21,6 +22,18 @@ import (
"gorm.io/gorm/clause"
)
var purchaseCreateFlights = struct {
sync.Mutex
calls map[string]*purchaseCreateCall
}{calls: make(map[string]*purchaseCreateCall)}
type purchaseCreateCall struct {
done chan struct{}
task models.PurchaseTask
replayed bool
err error
}
const DefaultLeaseDuration = 2 * time.Minute
type Service struct {
@@ -40,7 +53,122 @@ type SpecMatcher interface {
Resolve(context.Context, aimatching.MatchRequest) (aimatching.MatchResult, error)
}
type externalMatchPlan struct {
fingerprint string
result aimatching.MatchResult
}
type matchFingerprint struct {
SYBProductID uint64 `json:"sybProductId"`
TargetColor string `json:"targetColor"`
TargetSize string `json:"targetSize"`
Quantity int64 `json:"quantity"`
ShopeeProductID uint64 `json:"shopeeProductId"`
ShopeeSpecs json.RawMessage `json:"shopeeSpecs"`
PDDProductID uint64 `json:"pddProductId"`
PDDStatus string `json:"pddStatus"`
PDDSpecs json.RawMessage `json:"pddSpecs"`
Colors []string `json:"colors"`
Sizes []string `json:"sizes"`
}
func (s *Service) Create(ctx context.Context, req CreateRequest) (models.PurchaseTask, bool, error) {
purchaseCreateFlights.Lock()
if call := purchaseCreateFlights.calls[req.RequestID]; call != nil {
purchaseCreateFlights.Unlock()
select {
case <-ctx.Done():
return models.PurchaseTask{}, false, ctx.Err()
case <-call.done:
return call.task, true, call.err
}
}
call := &purchaseCreateCall{done: make(chan struct{})}
purchaseCreateFlights.calls[req.RequestID] = call
purchaseCreateFlights.Unlock()
call.task, call.replayed, call.err = s.create(ctx, req)
close(call.done)
purchaseCreateFlights.Lock()
delete(purchaseCreateFlights.calls, req.RequestID)
purchaseCreateFlights.Unlock()
return call.task, call.replayed, call.err
}
func (s *Service) prepareExternalMatch(ctx context.Context, req CreateRequest) (*externalMatchPlan, error) {
if req.ExecutionMode != models.PurchaseExecutionModeLive || req.SYBProductID == nil {
return nil, nil
}
var syb models.SYBProduct
if err := s.DB.WithContext(ctx).First(&syb, *req.SYBProductID).Error; err != nil {
return nil, notFound(err, "顺云宝商品不存在")
}
if syb.ShopeeProductID == nil {
return nil, fail(CodeInvalidRequest, "该商品尚未关联蝦皮商品")
}
var shopee models.ShopeeProduct
if err := s.DB.WithContext(ctx).First(&shopee, *syb.ShopeeProductID).Error; err != nil {
return nil, notFound(err, "蝦皮商品不存在")
}
if shopee.PDDProductID == nil {
return nil, fail(CodeInvalidRequest, "该蝦皮商品尚未关联拼多多商品")
}
var pdd models.PDDProduct
if err := s.DB.WithContext(ctx).First(&pdd, *shopee.PDDProductID).Error; err != nil {
return nil, notFound(err, "拼多多商品不存在")
}
targetColor, targetSize := strings.TrimSpace(req.TargetColor), strings.TrimSpace(req.TargetSize)
if targetColor == "" {
targetColor = syb.TargetColor
}
if targetSize == "" {
targetSize = syb.TargetSize
}
_, _, source := confirmedMappings(shopee.SpecsJSON, targetColor, targetSize)
candidates, usable := archiveCandidates(pdd.SpecsJSON, targetColor, targetSize)
if pdd.Status != "active" || source != "unresolved" || !usable {
return nil, nil
}
request := aimatching.MatchRequest{TargetColor: targetColor, TargetSize: targetSize, Colors: candidates.Colors, Sizes: candidates.Sizes}
if _, deterministic := aimatching.DeterministicMatch(request); deterministic {
return nil, nil
}
fingerprint, err := creationMatchFingerprint(syb, shopee, pdd, request)
if err != nil {
return nil, internal(err)
}
match, err := s.matcher().Resolve(ctx, request)
if err != nil {
return nil, purchaseMatchError(err)
}
return &externalMatchPlan{fingerprint: fingerprint, result: match}, nil
}
func creationMatchFingerprint(syb models.SYBProduct, shopee models.ShopeeProduct, pdd models.PDDProduct, request aimatching.MatchRequest) (string, error) {
canonical := func(raw string) (json.RawMessage, error) {
var value any
if err := json.Unmarshal([]byte(raw), &value); err != nil {
return nil, err
}
return json.Marshal(value)
}
shopeeSpecs, err := canonical(shopee.SpecsJSON)
if err != nil {
return "", err
}
pddSpecs, err := canonical(pdd.SpecsJSON)
if err != nil {
return "", err
}
payload, err := json.Marshal(matchFingerprint{SYBProductID: syb.ID, TargetColor: request.TargetColor, TargetSize: request.TargetSize, Quantity: syb.Quantity, ShopeeProductID: shopee.ID, ShopeeSpecs: shopeeSpecs, PDDProductID: pdd.ID, PDDStatus: pdd.Status, PDDSpecs: pddSpecs, Colors: request.Colors, Sizes: request.Sizes})
if err != nil {
return "", err
}
digest := sha256.Sum256(payload)
return hex.EncodeToString(digest[:]), nil
}
func (s *Service) create(ctx context.Context, req CreateRequest) (models.PurchaseTask, bool, error) {
if _, err := uuid.Parse(req.RequestID); err != nil {
return models.PurchaseTask{}, false, fail(CodeInvalidRequest, "requestId 无效")
}
@@ -59,6 +187,15 @@ func (s *Service) Create(ctx context.Context, req CreateRequest) (models.Purchas
return models.PurchaseTask{}, false, fail(CodeInvalidRequest, err.Error())
}
var out models.PurchaseTask
if err := s.DB.WithContext(ctx).Where("create_request_id = ?", req.RequestID).First(&out).Error; err == nil {
return out, true, nil
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
return models.PurchaseTask{}, false, internal(err)
}
externalPlan, err := s.prepareExternalMatch(ctx, req)
if err != nil {
return models.PurchaseTask{}, false, err
}
replayed := false
err = s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Where("create_request_id = ?", req.RequestID).First(&out).Error; err == nil {
@@ -199,6 +336,15 @@ func (s *Service) Create(ctx context.Context, req CreateRequest) (models.Purchas
}
candidates, archiveUsable := archiveCandidates(pdd.SpecsJSON, targetColor, targetSize)
matchRequest := aimatching.MatchRequest{TargetColor: targetColor, TargetSize: targetSize, Colors: candidates.Colors, Sizes: candidates.Sizes}
if externalPlan != nil {
fingerprint, fingerprintErr := creationMatchFingerprint(syb, shopee, pdd, matchRequest)
if fingerprintErr != nil {
return internal(fingerprintErr)
}
if fingerprint != externalPlan.fingerprint {
return fail(CodeMappingRequired, "规格匹配输入已变化，请重新创建采购任务")
}
}
if taskType == models.PurchaseTaskTypeStock {
// Direct stock selection was already validated above. It never enters
// mapping, deterministic matching, or the external AI fallback.
@@ -216,16 +362,26 @@ func (s *Service) Create(ctx context.Context, req CreateRequest) (models.Purchas
} else if !archiveUsable {
mappedColor, mappedSize, specSource = "", "", "unresolved"
} else {
match, matchErr := s.matcher().Resolve(ctx, matchRequest)
if matchErr != nil {
return purchaseMatchError(matchErr)
if externalPlan != nil {
// The complete input was revalidated above before selecting a branch.
mappedColor, mappedSize, specSource = externalPlan.result.MappedColor, externalPlan.result.MappedSize, externalPlan.result.Source
decision, marshalErr := json.Marshal(externalPlan.result.Decision)
if marshalErr != nil {
return internal(marshalErr)
}
decisionSnapshot = string(decision)
} else {
match, matchErr := s.matcher().Resolve(ctx, matchRequest)
if matchErr != nil {
return purchaseMatchError(matchErr)
}
mappedColor, mappedSize, specSource = match.MappedColor, match.MappedSize, match.Source
decision, marshalErr := json.Marshal(match.Decision)
if marshalErr != nil {
return internal(marshalErr)
}
decisionSnapshot = string(decision)
}
mappedColor, mappedSize, specSource = match.MappedColor, match.MappedSize, match.Source
decision, marshalErr := json.Marshal(match.Decision)
if marshalErr != nil {
return internal(marshalErr)
}
decisionSnapshot = string(decision)
}
if specSource == "unresolved" && !containsString(rule.RequiredCapabilities, purchasecontract.CapabilitySpecProbeV1) {
return fail(CodeMappingRequired, "规格映射不完整，所选规则不支持规格探测")
server/app/goauto/purchase/service_test.go
+104
View File
@@ -6,9 +6,11 @@ import (
"errors"
"fmt"
"strings"
"sync"
"testing"
"time"
"go-admin/app/goauto/aimatching"
"go-admin/app/goauto/device"
"go-admin/app/goauto/migrations"
"go-admin/app/goauto/models"
@@ -21,6 +23,34 @@ import (
"gorm.io/gorm/logger"
)
type blockingMatch struct {
started chan struct{}
release chan struct{}
calls int
mu sync.Mutex
}
func (matcher *blockingMatch) Resolve(ctx context.Context, request aimatching.MatchRequest) (aimatching.MatchResult, error) {
matcher.mu.Lock()
matcher.calls++
if matcher.calls == 1 {
close(matcher.started)
}
matcher.mu.Unlock()
select {
case <-ctx.Done():
return aimatching.MatchResult{}, ctx.Err()
case <-matcher.release:
return aimatching.RecordedMatch(request, aimatching.SourceAI, "白色", "XL", "唯一候选"), nil
}
}
func (matcher *blockingMatch) count() int {
matcher.mu.Lock()
defer matcher.mu.Unlock()
return matcher.calls
}
type fixture struct {
syb models.SYBProduct
shopee models.ShopeeProduct
@@ -391,6 +421,80 @@ func TestCreateRejectsUsableArchiveWhenNeitherMatcherFindsSpec(t *testing.T) {
}
}
func TestCreateWaitsForMatcherBeforeOpeningWriteTransactionAndDeduplicatesConcurrentRequest(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), false)
if err := db.Model(&models.PDDProduct{}).Where("id = ?", f.pdd.ID).Update("specs_json", `[{"name":"颜色","role":"color","values":[{"name":"白色","selectable":true,"priceCent":2000},{"name":"米白色","selectable":true,"priceCent":2000}]},{"name":"尺码","role":"size","values":[{"name":"XL","selectable":true}]}]`).Error; err != nil {
t.Fatal(err)
}
if err := db.Model(&models.SYBProduct{}).Where("id = ?", f.syb.ID).Update("target_color", "象牙白").Error; err != nil {
t.Fatal(err)
}
matcher := &blockingMatch{started: make(chan struct{}), release: make(chan struct{})}
service := testService(db)
service.Matcher = matcher
requestID := uuid.NewString()
request := CreateRequest{RequestID: requestID, ExecutionMode: models.PurchaseExecutionModeLive, SYBProductID: &f.syb.ID, DeviceID: &f.device.ID, MinUnitPriceCent: 400, MaxUnitPriceCent: 3000, RuleSnapshot: liveRule(true)}
type result struct {
task models.PurchaseTask
replayed bool
err error
}
results := make(chan result, 2)
go func() {
task, replayed, err := service.Create(context.Background(), request)
results <- result{task, replayed, err}
}()
<-matcher.started
if err := db.Model(&models.SYBProduct{}).Where("id = ?", f.syb.ID).Update("product_title", "等待期间可更新").Error; err != nil {
t.Fatalf("matcher wait held database transaction: %v", err)
}
go func() {
task, replayed, err := service.Create(context.Background(), request)
results <- result{task, replayed, err}
}()
close(matcher.release)
first, second := <-results, <-results
if first.err != nil || second.err != nil || first.task.ID == 0 || first.task.ID != second.task.ID {
t.Fatalf("concurrent result mismatch: first=%+v second=%+v", first, second)
}
if matcher.count() != 1 || (!first.replayed && !second.replayed) {
t.Fatalf("calls=%d replayed=%v/%v", matcher.count(), first.replayed, second.replayed)
}
}
func TestCreateRejectsExternalMatchWhenInputChanges(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), false)
if err := db.Model(&models.PDDProduct{}).Where("id = ?", f.pdd.ID).Update("specs_json", `[{"name":"颜色","role":"color","values":[{"name":"白色","selectable":true,"priceCent":2000},{"name":"米白色","selectable":true,"priceCent":2000}]},{"name":"尺码","role":"size","values":[{"name":"XL","selectable":true}]}]`).Error; err != nil {
t.Fatal(err)
}
if err := db.Model(&models.SYBProduct{}).Where("id = ?", f.syb.ID).Update("target_color", "象牙白").Error; err != nil {
t.Fatal(err)
}
matcher := &blockingMatch{started: make(chan struct{}), release: make(chan struct{})}
service := testService(db)
service.Matcher = matcher
done := make(chan error, 1)
go func() {
_, _, err := service.Create(context.Background(), CreateRequest{RequestID: uuid.NewString(), ExecutionMode: models.PurchaseExecutionModeLive, SYBProductID: &f.syb.ID, DeviceID: &f.device.ID, MinUnitPriceCent: 400, MaxUnitPriceCent: 3000, RuleSnapshot: liveRule(true)})
done <- err
}()
<-matcher.started
if err := db.Model(&models.PDDProduct{}).Where("id = ?", f.pdd.ID).Update("status", "disabled").Error; err != nil {
t.Fatal(err)
}
close(matcher.release)
if err := <-done; code(err) != CodeMappingRequired {
t.Fatalf("changed input code=%s err=%v", code(err), err)
}
var count int64
db.Model(&models.PurchaseTask{}).Count(&count)
if count != 0 {
t.Fatalf("stale match created %d tasks", count)
}
}
func TestOrderUnknownIsNotAutomaticallyRedispatched(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
web/src/api/goauto/purchase-tasks.js
+3 -2
View File
@@ -3,9 +3,10 @@ import request from '@/utils/request'
export function listPurchaseTasks(params) { return request({ url: '/api/admin/v1/purchase-tasks', method: 'get', params }) }
export function getPurchaseTask(taskId) { return request({ url: `/api/admin/v1/purchase-tasks/${taskId}`, method: 'get' }) }
export function previewPurchaseTasks(data, options = {}) { return request({ url: '/api/admin/v1/purchase-tasks/batch-preview', method: 'post', data, ...options }) }
export function createPurchaseTasksBatch(data) { return request({ url: '/api/admin/v1/purchase-tasks/batch', method: 'post', data }) }
export function retryPurchaseTasksBatch(data) { return request({ url: '/api/admin/v1/purchase-tasks/batch-retry', method: 'post', data }) }
export function createPurchaseTasksBatch(data) { return request({ url: '/api/admin/v1/purchase-tasks/batch', method: 'post', data, timeout: 60000 }) }
export function retryPurchaseTasksBatch(data) { return request({ url: '/api/admin/v1/purchase-tasks/batch-retry', method: 'post', data, timeout: 60000 }) }
export function createStockPurchaseTask(data, options = {}) { return request({ url: '/api/admin/v1/purchase-tasks/stock', method: 'post', data, ...options }) }
export function createPurchaseTask(data) { return request({ url: '/api/admin/v1/purchase-tasks', method: 'post', data, timeout: 60000 }) }
export function authorizeRepurchase(taskId, data) { return request({ url: `/api/admin/v1/purchase-tasks/${taskId}/authorize-repurchase`, method: 'post', data }) }
export function reviewPurchasePayment(taskId, data) { return request({ url: `/api/admin/v1/purchase-tasks/${taskId}/payment-review`, method: 'post', data }) }
export function selectPurchaseWriteback(taskId, data) { return request({ url: `/api/admin/v1/purchase-tasks/${taskId}/writeback-candidate`, method: 'post', data }) }