diff --git a/server/app/goauto/purchase/batch.go b/server/app/goauto/purchase/batch.go index 382d0e8..ffa2037 100644 --- a/server/app/goauto/purchase/batch.go +++ b/server/app/goauto/purchase/batch.go @@ -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] diff --git a/server/app/goauto/purchase/service.go b/server/app/goauto/purchase/service.go index 46e118d..40d545f 100644 --- a/server/app/goauto/purchase/service.go +++ b/server/app/goauto/purchase/service.go @@ -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, "规格映射不完整，所选规则不支持规格探测") diff --git a/server/app/goauto/purchase/service_test.go b/server/app/goauto/purchase/service_test.go index 305a556..bbc7a60 100644 --- a/server/app/goauto/purchase/service_test.go +++ b/server/app/goauto/purchase/service_test.go @@ -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) diff --git a/web/src/api/goauto/purchase-tasks.js b/web/src/api/goauto/purchase-tasks.js index 8c6bb5e..3f80349 100644 --- a/web/src/api/goauto/purchase-tasks.js +++ b/web/src/api/goauto/purchase-tasks.js @@ -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 }) }