diff --git a/backend-api/internal/config/config.go b/backend-api/internal/config/config.go index 32ac8d5..57d9b6e 100644 --- a/backend-api/internal/config/config.go +++ b/backend-api/internal/config/config.go @@ -201,7 +201,7 @@ func Load(lookup LookupEnvironment) (Config, error) { TLSPrivateKey: cleanOptionalPath(tlsPrivateKey), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, - WriteTimeout: 30 * time.Second, + WriteTimeout: 70 * time.Second, IdleTimeout: 60 * time.Second, ShutdownTimeout: 10 * time.Second, MaxHeaderBytes: 1 << 20, diff --git a/backend-api/internal/config/config_test.go b/backend-api/internal/config/config_test.go index 487187a..ce302d3 100644 --- a/backend-api/internal/config/config_test.go +++ b/backend-api/internal/config/config_test.go @@ -29,6 +29,9 @@ func TestLoadUsesSafeDefaults(t *testing.T) { cfg.MaxHeaderBytes <= 0 { t.Fatal("server safety limits must all be positive") } + if cfg.WriteTimeout != 70*time.Second { + t.Fatalf("WriteTimeout = %s", cfg.WriteTimeout) + } if cfg.ClaimLease != 10*time.Minute || cfg.RunningLease != 30*time.Minute || cfg.ReadinessTTL != 2*time.Minute { diff --git a/backend-api/internal/transport/httpapi/admin_handlers.go b/backend-api/internal/transport/httpapi/admin_handlers.go index 69dfee5..09bd838 100644 --- a/backend-api/internal/transport/httpapi/admin_handlers.go +++ b/backend-api/internal/transport/httpapi/admin_handlers.go @@ -852,6 +852,8 @@ func writeUsecaseError(ctx *gin.Context, err error) { status = http.StatusBadGateway case "OCR_SERVICE_INVALID", "ERP_UNAVAILABLE": status = http.StatusServiceUnavailable + case "FREIGHT_SYNC_TIMEOUT": + status = http.StatusGatewayTimeout } message := typed.Message if preflightMessage, ok := freightPreflightPublicMessage(typed.Code); ok { @@ -883,6 +885,10 @@ func freightPreflightPublicMessage(code string) (string, bool) { return "OCR service is invalid", true case "ERP_UNAVAILABLE": return "ERP is temporarily unavailable", true + case "FREIGHT_SYNC_TIMEOUT": + return "freight order sync exceeded 55 seconds", true + case "FREIGHT_SYNC_BUSY": + return "another freight order sync is already running", true default: return "", false } diff --git a/backend-api/internal/transport/httpapi/admin_handlers_test.go b/backend-api/internal/transport/httpapi/admin_handlers_test.go index 39102b3..60fe2c2 100644 --- a/backend-api/internal/transport/httpapi/admin_handlers_test.go +++ b/backend-api/internal/transport/httpapi/admin_handlers_test.go @@ -405,7 +405,9 @@ func TestAdminFreightAPIImportsAllItemsWithoutPII(t *testing.T) { ), "freight-sync-1", ) - if create.Code != http.StatusAccepted { + if create.Code != http.StatusCreated || + create.Header().Get("Location") != "/api/v1/freight-orders" || + !strings.Contains(create.Body.String(), `"status":"SUCCEEDED"`) { t.Fatalf("create status/body = %d / %s", create.Code, create.Body) } var createBody struct { @@ -486,8 +488,9 @@ func TestAdminFreightAPIImportsAllItemsWithoutPII(t *testing.T) { ), "freight-sync-1", ) - if replay.Code != http.StatusAccepted || + if replay.Code != http.StatusOK || !strings.Contains(replay.Body.String(), `"replayed":true`) || + !strings.Contains(replay.Body.String(), `"status":"SUCCEEDED"`) || !strings.Contains(replay.Body.String(), createBody.Sync.ID) { t.Fatalf("replay status/body = %d / %s", replay.Code, replay.Body) } @@ -515,6 +518,9 @@ func TestAdminFreightDateSyncAdvancesInspectableWatermark(t *testing.T) { "freight-date-sync-1", ) requireAdminStatus(t, create, http.StatusAccepted) + if !strings.Contains(create.Body.String(), `"status":"PENDING"`) { + t.Fatalf("date sync create response = %s", create.Body) + } var created struct { Sync struct { ID string `json:"id"` @@ -589,31 +595,9 @@ func TestAdminProcurementAPIProducesImmutablePendingTask(t *testing.T) { ), "procurement-freight-sync", ) - var syncBody struct { - Sync struct { - ID string `json:"id"` - } `json:"sync"` - } - decodeResponse(t, createSync, &syncBody) - succeeded := false - for attempt := 0; attempt < 50; attempt++ { - status := performAdminRequest( - t, - fixture.router, - http.MethodGet, - "/api/v1/freight-syncs/"+syncBody.Sync.ID, - "", - nil, - "", - ) - if strings.Contains(status.Body.String(), `"status":"SUCCEEDED"`) { - succeeded = true - break - } - time.Sleep(10 * time.Millisecond) - } - if !succeeded { - t.Fatal("freight sync did not succeed") + requireAdminStatus(t, createSync, http.StatusCreated) + if !strings.Contains(createSync.Body.String(), `"status":"SUCCEEDED"`) { + t.Fatalf("freight sync response = %s", createSync.Body) } orders := performAdminRequest( t, diff --git a/backend-api/internal/transport/httpapi/freight_handlers.go b/backend-api/internal/transport/httpapi/freight_handlers.go index 584be19..971e528 100644 --- a/backend-api/internal/transport/httpapi/freight_handlers.go +++ b/backend-api/internal/transport/httpapi/freight_handlers.go @@ -106,7 +106,15 @@ func (h *adminHandlers) createFreightSync(ctx *gin.Context) { return } ctx.Header("Cache-Control", "no-store") - ctx.JSON(http.StatusAccepted, gin.H{ + status := http.StatusAccepted + if request.Mode == domain.FreightSyncOrderNumber { + status = http.StatusCreated + if result.Replayed { + status = http.StatusOK + } + ctx.Header("Location", "/api/v1/freight-orders") + } + ctx.JSON(status, gin.H{ "sync": freightSyncResponse(result.Run), "replayed": result.Replayed, }) diff --git a/backend-api/internal/transport/httpapi/preflight_error_test.go b/backend-api/internal/transport/httpapi/preflight_error_test.go index 15f411d..a780987 100644 --- a/backend-api/internal/transport/httpapi/preflight_error_test.go +++ b/backend-api/internal/transport/httpapi/preflight_error_test.go @@ -21,6 +21,7 @@ func TestWriteUsecaseErrorUsesPreflightStatusAndCode(t *testing.T) { {"ERP_RESPONSE_INVALID", http.StatusBadGateway}, {"OCR_SERVICE_INVALID", http.StatusServiceUnavailable}, {"ERP_UNAVAILABLE", http.StatusServiceUnavailable}, + {"FREIGHT_SYNC_TIMEOUT", http.StatusGatewayTimeout}, } for _, testCase := range testCases { t.Run(testCase.code, func(t *testing.T) { diff --git a/backend-api/internal/transport/webui/handler.go b/backend-api/internal/transport/webui/handler.go index e522d30..a290783 100644 --- a/backend-api/internal/transport/webui/handler.go +++ b/backend-api/internal/transport/webui/handler.go @@ -13,6 +13,7 @@ import ( "time" "unicode/utf8" + "cmroubao/backend-api/internal/domain" "cmroubao/backend-api/internal/transport/authcommon" "github.com/gin-gonic/gin" @@ -319,6 +320,7 @@ func (h *Handler) ListFreight(ctx *gin.Context) { CSRFToken: token, }, Orders: orders, + Notice: freightNotice(ctx.Query("notice")), }) } @@ -443,6 +445,10 @@ func (h *Handler) CreateFreightImport(ctx *gin.Context) { }) return } + if mode == domain.FreightSyncOrderNumber && run.Status == "SUCCEEDED" { + ctx.Redirect(http.StatusSeeOther, "/freight?notice=import-succeeded") + return + } ctx.Redirect(http.StatusSeeOther, "/freight/import?sync="+pathEscape(run.ID)) } @@ -452,7 +458,7 @@ func (h *Handler) logFreightImportFailure(code string, status int) { return } h.logEvent( - "freight_import_preflight_failed code=" + code + + "freight_import_failed code=" + code + " status=" + strconv.Itoa(status), ) } @@ -463,12 +469,20 @@ func freightImportError(err error) (string, string, string) { return "OCR_SERVICE_INVALID", "OCR 服务无效", "OCR 服务无效，请检查本机 OCR 服务和 CMROUBAO_OCR_API_URL 后重试。" case errors.Is(err, ErrERPNotConfigured): return "ERP_NOT_CONFIGURED", "ERP 凭证未配置", "ERP 账号或密码未配置，请检查 backend-api/.env 后重试。" + case errors.Is(err, ErrERPSessionNeeded): + return "ERP_SESSION_REQUIRED", "ERP 会话已失效", "ERP 会话已失效，请使用相同提交标识重试。" + case errors.Is(err, ErrERPFreightNotFound): + return "ERP_FREIGHT_NOT_FOUND", "未找到货运单", "ERP 中没有找到该完整单号，请检查后重试。" case errors.Is(err, ErrERPLoginRejected): return "ERP_LOGIN_REJECTED", "ERP 登录被拒绝", "请检查 ERP 账号密码及 OCR 识别结果后重试。" case errors.Is(err, ErrERPProtocol): return "ERP_RESPONSE_INVALID", "ERP 响应无效", "ERP 返回格式无法确认，请稍后重试。" case errors.Is(err, ErrERPUnavailable): return "ERP_UNAVAILABLE", "ERP 暂时不可用", "ERP 服务暂时不可用，请稍后使用相同提交标识重试。" + case errors.Is(err, ErrFreightSyncBusy): + return "FREIGHT_SYNC_BUSY", "ERP 正在同步", "已有完整单号正在同步，请等待完成后重试。" + case errors.Is(err, ErrFreightSyncTimeout): + return "FREIGHT_SYNC_TIMEOUT", "ERP 同步超时", "完整单号同步超过 55 秒，请使用相同提交标识重试。" default: return "", "", "同步任务创建失败，请稍后使用相同提交标识重试。" } @@ -646,6 +660,8 @@ func freightNotice(value string) string { return "参考图已绑定，可以生成采购任务。" case "task-conflict": return "需求状态或来源已变化，当前不能生成任务。" + case "import-succeeded": + return "ERP 货运单已同步，货运信息和商品明细已更新。" default: return "" } @@ -1270,10 +1286,16 @@ func serviceErrorStatus(err error) int { return http.StatusServiceUnavailable case errors.Is(err, ErrERPNotConfigured), errors.Is(err, ErrERPLoginRejected): return http.StatusUnprocessableEntity + case errors.Is(err, ErrERPFreightNotFound): + return http.StatusNotFound case errors.Is(err, ErrERPProtocol): return http.StatusBadGateway - case errors.Is(err, ErrERPUnavailable): + case errors.Is(err, ErrERPSessionNeeded), errors.Is(err, ErrERPUnavailable): return http.StatusServiceUnavailable + case errors.Is(err, ErrFreightSyncBusy): + return http.StatusConflict + case errors.Is(err, ErrFreightSyncTimeout): + return http.StatusGatewayTimeout case errors.Is(err, ErrConflict): return http.StatusConflict case errors.Is(err, context.DeadlineExceeded): @@ -1336,6 +1358,7 @@ type pageView struct { type freightPage struct { Page pageView Orders []FreightOrder + Notice string } type freightImportPage struct { diff --git a/backend-api/internal/transport/webui/handler_test.go b/backend-api/internal/transport/webui/handler_test.go index 17332e2..0bc22fd 100644 --- a/backend-api/internal/transport/webui/handler_test.go +++ b/backend-api/internal/transport/webui/handler_test.go @@ -923,7 +923,7 @@ func TestRendererUsesMissingKeyErrors(t *testing.T) { } } -func TestFreightPagesEscapeSourceDataAndCreateAsyncSync(t *testing.T) { +func TestFreightPagesEscapeSourceDataAndCreateSynchronousOrderSync(t *testing.T) { now := time.Date(2026, 7, 28, 3, 4, 5, 0, time.UTC) service := &fakeFreightService{ fakeService: &fakeService{}, @@ -938,7 +938,7 @@ func TestFreightPagesEscapeSourceDataAndCreateAsyncSync(t *testing.T) { }}, createResult: FreightSync{ ID: testTaskID, - Status: "PENDING", + Status: "SUCCEEDED", CreatedAt: now, }, } @@ -987,7 +987,7 @@ func TestFreightPagesEscapeSourceDataAndCreateAsyncSync(t *testing.T) { router.ServeHTTP(response, request) if response.Code != http.StatusSeeOther || response.Header().Get("Location") != - "/freight/import?sync="+testTaskID { + "/freight?notice=import-succeeded" { t.Fatalf( "create sync status/location = %d / %q", response.Code, @@ -998,6 +998,17 @@ func TestFreightPagesEscapeSourceDataAndCreateAsyncSync(t *testing.T) { service.createInput.IdempotencyKey != idempotencyKey { t.Fatalf("create input = %+v", service.createInput) } + list = performRequest( + t, + router, + http.MethodGet, + "/freight?notice=import-succeeded", + nil, + "", + ) + if !strings.Contains(list.Body.String(), "ERP 货运单已同步") { + t.Fatalf("freight success notice = %s", list.Body) + } } func TestFreightDateFormSubmitsManualRange(t *testing.T) { @@ -1042,6 +1053,8 @@ func TestFreightDateFormSubmitsManualRange(t *testing.T) { router.ServeHTTP(response, request) if response.Code != http.StatusSeeOther || + response.Header().Get("Location") != + "/freight/import?sync="+testTaskID || service.createInput.Mode != "CREATED_RANGE" || service.createInput.CreatedFrom != "2026-07-22" || service.createInput.CreatedTo != "2026-07-28" { @@ -1116,9 +1129,13 @@ func TestFreightImportShowsStablePreflightErrorDialog(t *testing.T) { }{ {"ocr", ErrOCRServiceInvalid, http.StatusServiceUnavailable, "OCR_SERVICE_INVALID", "OCR 服务无效"}, {"not configured", ErrERPNotConfigured, http.StatusUnprocessableEntity, "ERP_NOT_CONFIGURED", "ERP 凭证未配置"}, + {"session", ErrERPSessionNeeded, http.StatusServiceUnavailable, "ERP_SESSION_REQUIRED", "ERP 会话已失效"}, + {"not found", ErrERPFreightNotFound, http.StatusNotFound, "ERP_FREIGHT_NOT_FOUND", "未找到货运单"}, {"login rejected", ErrERPLoginRejected, http.StatusUnprocessableEntity, "ERP_LOGIN_REJECTED", "ERP 登录被拒绝"}, {"protocol", ErrERPProtocol, http.StatusBadGateway, "ERP_RESPONSE_INVALID", "ERP 响应无效"}, {"unavailable", ErrERPUnavailable, http.StatusServiceUnavailable, "ERP_UNAVAILABLE", "ERP 暂时不可用"}, + {"busy", ErrFreightSyncBusy, http.StatusConflict, "FREIGHT_SYNC_BUSY", "ERP 正在同步"}, + {"timeout", ErrFreightSyncTimeout, http.StatusGatewayTimeout, "FREIGHT_SYNC_TIMEOUT", "ERP 同步超时"}, } for _, testCase := range testCases { t.Run(testCase.name, func(t *testing.T) { @@ -1151,7 +1168,7 @@ func TestFreightImportShowsStablePreflightErrorDialog(t *testing.T) { service.createInput.OrderNumber != "ORDER-123" { t.Fatalf("dialog response/input = %d / %s / %+v", response.Code, response.Body, service.createInput) } - wantEvent := "freight_import_preflight_failed code=" + + wantEvent := "freight_import_failed code=" + testCase.code + " status=" + strconv.Itoa(testCase.status) if event != wantEvent || strings.Contains(event, "private") { t.Fatalf("event = %q, want %q", event, wantEvent) diff --git a/backend-api/internal/transport/webui/preflight_error_test.go b/backend-api/internal/transport/webui/preflight_error_test.go index ae4a330..e4a5e93 100644 --- a/backend-api/internal/transport/webui/preflight_error_test.go +++ b/backend-api/internal/transport/webui/preflight_error_test.go @@ -5,6 +5,7 @@ import ( "testing" "cmroubao/backend-api/internal/domain" + "cmroubao/backend-api/internal/usecase" ) func TestMapFreightPreflightErrorUsesStablePublicErrors(t *testing.T) { @@ -14,6 +15,8 @@ func TestMapFreightPreflightErrorUsesStablePublicErrors(t *testing.T) { }{ {domain.ErrFreightSourceOCRInvalid, ErrOCRServiceInvalid}, {domain.ErrFreightSourceNotConfigured, ErrERPNotConfigured}, + {domain.ErrFreightSourceSessionNeeded, ErrERPSessionNeeded}, + {domain.ErrFreightSourceNotFound, ErrERPFreightNotFound}, {domain.ErrFreightSourceLoginRejected, ErrERPLoginRejected}, {domain.ErrFreightSourceProtocol, ErrERPProtocol}, {domain.ErrFreightSourceUnavailable, ErrERPUnavailable}, @@ -33,6 +36,8 @@ func TestMapFreightCreateErrorPreservesPreflightErrors(t *testing.T) { }{ {domain.ErrFreightSourceOCRInvalid, ErrOCRServiceInvalid}, {domain.ErrFreightSourceNotConfigured, ErrERPNotConfigured}, + {domain.ErrFreightSourceSessionNeeded, ErrERPSessionNeeded}, + {domain.ErrFreightSourceNotFound, ErrERPFreightNotFound}, {domain.ErrFreightSourceLoginRejected, ErrERPLoginRejected}, {domain.ErrFreightSourceProtocol, ErrERPProtocol}, {domain.ErrFreightSourceUnavailable, ErrERPUnavailable}, @@ -43,3 +48,23 @@ func TestMapFreightCreateErrorPreservesPreflightErrors(t *testing.T) { } } } + +func TestMapFreightCreateErrorUsesStableSyncErrors(t *testing.T) { + for _, testCase := range []struct { + code string + want error + }{ + {"FREIGHT_SYNC_BUSY", ErrFreightSyncBusy}, + {"FREIGHT_SYNC_TIMEOUT", ErrFreightSyncTimeout}, + } { + source := &usecase.Error{ + Kind: usecase.ErrorKindUnavailable, + Code: testCase.code, + Fields: map[string]string{}, + } + actual := mapFreightCreateError(source) + if !errors.Is(actual, testCase.want) || !errors.Is(actual, source) { + t.Fatalf("mapFreightCreateError(%s) = %v", testCase.code, actual) + } + } +} diff --git a/backend-api/internal/transport/webui/templates/freight-import.gohtml b/backend-api/internal/transport/webui/templates/freight-import.gohtml index b7f117d..d728ad7 100644 --- a/backend-api/internal/transport/webui/templates/freight-import.gohtml +++ b/backend-api/internal/transport/webui/templates/freight-import.gohtml @@ -63,7 +63,7 @@ - +
diff --git a/backend-api/internal/transport/webui/templates/freight.gohtml b/backend-api/internal/transport/webui/templates/freight.gohtml index 7882cc7..300bf0c 100644 --- a/backend-api/internal/transport/webui/templates/freight.gohtml +++ b/backend-api/internal/transport/webui/templates/freight.gohtml @@ -15,6 +15,7 @@ 导入货运单 + {{if .Notice}}
{{.Notice}}
{{end}}
{{if .Orders}} diff --git a/backend-api/internal/transport/webui/types.go b/backend-api/internal/transport/webui/types.go index 23a7b43..bb3adec 100644 --- a/backend-api/internal/transport/webui/types.go +++ b/backend-api/internal/transport/webui/types.go @@ -8,19 +8,22 @@ import ( ) var ( - ErrNotFound = errors.New("resource not found") - ErrForbidden = errors.New("resource forbidden") - ErrConflict = errors.New("resource conflict") - ErrValidation = errors.New("validation failed") - ErrInvalidFile = errors.New("invalid file") - ErrUnavailable = errors.New("service unavailable") - ErrERPNotConfigured = errors.New("ERP is not configured") - ErrERPSessionNeeded = errors.New("ERP session is required") - ErrERPCaptchaInvalid = errors.New("ERP captcha is invalid") - ErrERPLoginRejected = errors.New("ERP login was rejected") - ErrERPProtocol = errors.New("ERP protocol is invalid") - ErrERPUnavailable = errors.New("ERP is unavailable") - ErrOCRServiceInvalid = errors.New("OCR service is invalid") + ErrNotFound = errors.New("resource not found") + ErrForbidden = errors.New("resource forbidden") + ErrConflict = errors.New("resource conflict") + ErrValidation = errors.New("validation failed") + ErrInvalidFile = errors.New("invalid file") + ErrUnavailable = errors.New("service unavailable") + ErrERPNotConfigured = errors.New("ERP is not configured") + ErrERPSessionNeeded = errors.New("ERP session is required") + ErrERPCaptchaInvalid = errors.New("ERP captcha is invalid") + ErrERPLoginRejected = errors.New("ERP login was rejected") + ErrERPProtocol = errors.New("ERP protocol is invalid") + ErrERPUnavailable = errors.New("ERP is unavailable") + ErrERPFreightNotFound = errors.New("ERP freight order was not found") + ErrOCRServiceInvalid = errors.New("OCR service is invalid") + ErrFreightSyncBusy = errors.New("freight sync is busy") + ErrFreightSyncTimeout = errors.New("freight sync timed out") ) // Service is the application boundary required by the server-rendered admin UI. diff --git a/backend-api/internal/transport/webui/usecase_adapter.go b/backend-api/internal/transport/webui/usecase_adapter.go index 0c9b79f..e9bbb90 100644 --- a/backend-api/internal/transport/webui/usecase_adapter.go +++ b/backend-api/internal/transport/webui/usecase_adapter.go @@ -334,6 +334,10 @@ func mapFreightPreflightError(err error) error { public = ErrOCRServiceInvalid case errors.Is(err, domain.ErrFreightSourceNotConfigured): public = ErrERPNotConfigured + case errors.Is(err, domain.ErrFreightSourceSessionNeeded): + public = ErrERPSessionNeeded + case errors.Is(err, domain.ErrFreightSourceNotFound): + public = ErrERPFreightNotFound case errors.Is(err, domain.ErrFreightSourceLoginRejected): public = ErrERPLoginRejected case errors.Is(err, domain.ErrFreightSourceProtocol): @@ -388,6 +392,19 @@ func mapFreightCreateError(err error) error { if mapped := mapFreightPreflightError(err); mapped != nil { return mapped } + var typed *usecase.Error + if errors.As(err, &typed) { + var public error + switch typed.Code { + case "FREIGHT_SYNC_BUSY": + public = ErrFreightSyncBusy + case "FREIGHT_SYNC_TIMEOUT": + public = ErrFreightSyncTimeout + } + if public != nil { + return &adapterError{public: public, cause: err} + } + } return mapUsecaseError(err) } diff --git a/backend-api/internal/usecase/freight_service.go b/backend-api/internal/usecase/freight_service.go index a8812fb..139924d 100644 --- a/backend-api/internal/usecase/freight_service.go +++ b/backend-api/internal/usecase/freight_service.go @@ -20,14 +20,19 @@ const ( maxFreightItemsPerOrder = 1000 maxFreightWindowDays = 7 freightWatermarkOverlap = 10 * time.Minute + orderFreightSyncTimeout = 55 * time.Second + freightCleanupTimeout = 5 * time.Second ) type FreightService struct { - repository FreightRepository - source FreightSource - clock Clock - ids IDGenerator - timeout time.Duration + repository FreightRepository + source FreightSource + clock Clock + ids IDGenerator + timeout time.Duration + orderTimeout time.Duration + cleanupTimeout time.Duration + orderSyncGate chan struct{} } type FreightSourcePreflight interface { @@ -67,11 +72,14 @@ func NewFreightService( return nil, errors.New("freight service dependencies are required") } return &FreightService{ - repository: repository, - source: source, - clock: clock, - ids: ids, - timeout: timeout, + repository: repository, + source: source, + clock: clock, + ids: ids, + timeout: timeout, + orderTimeout: orderFreightSyncTimeout, + cleanupTimeout: freightCleanupTimeout, + orderSyncGate: make(chan struct{}, 1), }, nil } @@ -105,7 +113,16 @@ func (service *FreightService) CreateOrderSync( fields, ) } - if err := service.ensureSource(ctx); err != nil { + if !service.acquireOrderSync() { + return CreateFreightSyncResult{}, freightSyncBusyError() + } + defer service.releaseOrderSync() + syncCtx, cancel := context.WithTimeout(ctx, service.orderTimeout) + defer cancel() + if err := service.ensureSource(syncCtx); err != nil { + if syncErr := freightContextError(syncCtx); syncErr != nil { + return CreateFreightSyncResult{}, syncErr + } return CreateFreightSyncResult{}, err } runID, err := service.ids.NewID() @@ -121,7 +138,7 @@ func (service *FreightService) CreateOrderSync( OrderNumber string `json:"order_number"` }{command.OrderNumber}) run, created, err := service.repository.CreateFreightSync( - ctx, + syncCtx, domain.FreightSyncRun{ ID: runID, CreatorSubject: command.CreatorSubject, @@ -138,10 +155,31 @@ func (service *FreightService) CreateOrderSync( if err != nil { return CreateFreightSyncResult{}, wrapRepositoryError(err) } - if created { - go service.execute(run) + if !created { + if run.Status != domain.FreightSyncSucceeded && + run.Status != domain.FreightSyncFailed { + return CreateFreightSyncResult{}, freightSyncBusyError() + } + return CreateFreightSyncResult{Run: run, Replayed: true}, nil } - return CreateFreightSyncResult{Run: run, Replayed: !created}, nil + completed, err := service.executeRun(syncCtx, run) + if err != nil { + return CreateFreightSyncResult{}, err + } + return CreateFreightSyncResult{Run: completed}, nil +} + +func (service *FreightService) acquireOrderSync() bool { + select { + case service.orderSyncGate <- struct{}{}: + return true + default: + return false + } +} + +func (service *FreightService) releaseOrderSync() { + <-service.orderSyncGate } func (service *FreightService) CreateDateSync( @@ -310,10 +348,24 @@ func (service *FreightService) ensureSource(ctx context.Context) error { func (service *FreightService) execute(run domain.FreightSyncRun) { ctx, cancel := context.WithTimeout(context.Background(), service.timeout) defer cancel() + _, _ = service.executeRun(ctx, run) +} + +func (service *FreightService) executeRun( + ctx context.Context, + run domain.FreightSyncRun, +) (domain.FreightSyncRun, error) { now := service.clock.Now().UTC() if err := service.repository.StartFreightSync(ctx, run.ID, now); err != nil { - return + code := freightFailureCode(ctx, "STORAGE_UNAVAILABLE") + return domain.FreightSyncRun{}, service.finishFreightFailure( + run.ID, + code, + wrapRepositoryError(err), + ) } + run.Status = domain.FreightSyncRunning + run.StartedAt = &now var source domain.FreightSourceBatch var err error if run.Mode == domain.FreightSyncCreatedRange { @@ -322,23 +374,21 @@ func (service *FreightService) execute(run domain.FreightSyncRun) { source, err = service.source.QueryOrder(ctx, run.OrderNumber) } if err != nil { - _ = service.repository.FailFreightSync( - ctx, + code := freightFailureCode(ctx, freightSourceErrorCode(err)) + return domain.FreightSyncRun{}, service.finishFreightFailure( run.ID, - freightSourceErrorCode(err), - service.clock.Now().UTC(), + code, + freightExecutionError(code, err), ) - return } batch, err := service.normalize(source) if err != nil { - _ = service.repository.FailFreightSync( - ctx, + code := freightFailureCode(ctx, "ERP_RESPONSE_INVALID") + return domain.FreightSyncRun{}, service.finishFreightFailure( run.ID, - "ERP_RESPONSE_INVALID", - service.clock.Now().UTC(), + code, + freightExecutionError(code, domain.ErrFreightSourceProtocol), ) - return } finishedAt := service.clock.Now().UTC() if run.Mode == domain.FreightSyncCreatedRange && @@ -359,13 +409,107 @@ func (service *FreightService) execute(run domain.FreightSyncRun) { ) } if err != nil { - _ = service.repository.FailFreightSync( - ctx, + code := freightFailureCode(ctx, "STORAGE_UNAVAILABLE") + return domain.FreightSyncRun{}, service.finishFreightFailure( run.ID, - "STORAGE_UNAVAILABLE", - service.clock.Now().UTC(), + code, + wrapRepositoryError(err), ) } + itemCount := 0 + for _, order := range batch.Orders { + itemCount += len(order.Items) + } + run.Status = domain.FreightSyncSucceeded + run.OrderCount = len(batch.Orders) + run.ItemCount = itemCount + run.FinishedAt = &finishedAt + return run, nil +} + +func (service *FreightService) finishFreightFailure( + runID, code string, + cause error, +) error { + timeout := service.cleanupTimeout + if timeout <= 0 { + timeout = freightCleanupTimeout + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + if err := service.repository.FailFreightSync( + ctx, + runID, + code, + service.clock.Now().UTC(), + ); err != nil { + return wrapRepositoryError(err) + } + return cause +} + +func freightFailureCode(ctx context.Context, fallback string) string { + switch ctx.Err() { + case context.DeadlineExceeded: + return "FREIGHT_SYNC_TIMEOUT" + case context.Canceled: + return "FREIGHT_SYNC_CANCELED" + default: + return fallback + } +} + +func freightContextError(ctx context.Context) error { + switch ctx.Err() { + case context.DeadlineExceeded: + return freightExecutionError( + "FREIGHT_SYNC_TIMEOUT", + context.DeadlineExceeded, + ) + case context.Canceled: + return freightExecutionError( + "FREIGHT_SYNC_CANCELED", + context.Canceled, + ) + default: + return nil + } +} + +func freightSyncBusyError() error { + result := newError( + ErrorKindConflict, + "FREIGHT_SYNC_BUSY", + "another freight order sync is already running", + nil, + ) + result.Retryable = true + return result +} + +func freightExecutionError(code string, cause error) error { + var kind ErrorKind + var message string + switch code { + case "FREIGHT_SYNC_TIMEOUT": + kind = ErrorKindUnavailable + message = "freight order sync exceeded its time limit" + case "FREIGHT_SYNC_CANCELED": + kind = ErrorKindUnavailable + message = "freight order sync was canceled" + case "ERP_FREIGHT_NOT_FOUND": + kind = ErrorKindNotFound + message = "ERP freight order was not found" + default: + kind = ErrorKindUnavailable + message = freightPreflightMessage(code) + } + result := newError(kind, code, message, cause) + result.Retryable = code == "FREIGHT_SYNC_TIMEOUT" || + code == "FREIGHT_SYNC_CANCELED" || + code == "ERP_UNAVAILABLE" || + code == "OCR_SERVICE_INVALID" + return result } func (service *FreightService) queryCreatedRange( diff --git a/backend-api/internal/usecase/freight_service_test.go b/backend-api/internal/usecase/freight_service_test.go index 4feaf34..0f8a626 100644 --- a/backend-api/internal/usecase/freight_service_test.go +++ b/backend-api/internal/usecase/freight_service_test.go @@ -3,6 +3,7 @@ package usecase import ( "context" "errors" + "sync" "testing" "time" @@ -98,6 +99,176 @@ func TestCreateFreightOrderSyncStopsBeforePersistingWhenOCRIsInvalid(t *testing. } } +func TestCreateFreightOrderSyncReturnsCommittedResult(t *testing.T) { + repository := &syncTrackingRepository{} + service, err := NewFreightService( + repository, + &recordingDateSource{}, + fakeClock{}, + &sequenceIDs{}, + time.Minute, + ) + if err != nil { + t.Fatalf("NewFreightService() error = %v", err) + } + if service.orderTimeout != 55*time.Second { + t.Fatalf("order timeout = %s", service.orderTimeout) + } + result, err := service.CreateOrderSync( + context.Background(), + validFreightOrderSyncCommand("sync-success"), + ) + if err != nil { + t.Fatalf("CreateOrderSync() error = %v", err) + } + if result.Replayed || result.Run.Status != domain.FreightSyncSucceeded || + result.Run.OrderCount != 1 || result.Run.ItemCount != 1 || + result.Run.StartedAt == nil || result.Run.FinishedAt == nil { + t.Fatalf("CreateOrderSync() result = %+v", result) + } + repository.mu.Lock() + defer repository.mu.Unlock() + if repository.startCalls != 1 || repository.completeCalls != 1 || + repository.failCalls != 0 { + t.Fatalf( + "start/complete/fail calls = %d/%d/%d", + repository.startCalls, + repository.completeCalls, + repository.failCalls, + ) + } +} + +func TestCreateFreightOrderSyncTimeoutUsesLiveCleanupContext(t *testing.T) { + repository := &syncTrackingRepository{} + source := &blockingOrderSource{} + service, err := NewFreightService( + repository, + source, + fakeClock{}, + &sequenceIDs{}, + time.Minute, + ) + if err != nil { + t.Fatalf("NewFreightService() error = %v", err) + } + service.orderTimeout = 10 * time.Millisecond + _, err = service.CreateOrderSync( + context.Background(), + validFreightOrderSyncCommand("sync-timeout"), + ) + assertUsecaseError(t, err, ErrorKindUnavailable, "FREIGHT_SYNC_TIMEOUT") + repository.mu.Lock() + defer repository.mu.Unlock() + if repository.failCalls != 1 || + repository.failCode != "FREIGHT_SYNC_TIMEOUT" || + repository.failContextErr != nil { + t.Fatalf( + "failure calls/code/context = %d/%q/%v", + repository.failCalls, + repository.failCode, + repository.failContextErr, + ) + } +} + +func TestCreateFreightOrderSyncCancellationUsesLiveCleanupContext(t *testing.T) { + repository := &syncTrackingRepository{} + source := &blockingOrderSource{started: make(chan struct{})} + service, err := NewFreightService( + repository, + source, + fakeClock{}, + &sequenceIDs{}, + time.Minute, + ) + if err != nil { + t.Fatalf("NewFreightService() error = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { + _, runErr := service.CreateOrderSync( + ctx, + validFreightOrderSyncCommand("sync-canceled"), + ) + result <- runErr + }() + select { + case <-source.started: + case <-time.After(time.Second): + t.Fatal("order sync did not reach source") + } + cancel() + err = <-result + assertUsecaseError(t, err, ErrorKindUnavailable, "FREIGHT_SYNC_CANCELED") + repository.mu.Lock() + defer repository.mu.Unlock() + if repository.failCalls != 1 || + repository.failCode != "FREIGHT_SYNC_CANCELED" || + repository.failContextErr != nil { + t.Fatalf( + "failure calls/code/context = %d/%q/%v", + repository.failCalls, + repository.failCode, + repository.failContextErr, + ) + } +} + +func TestCreateFreightOrderSyncRejectsConcurrentRequest(t *testing.T) { + repository := &syncTrackingRepository{} + source := &blockingOrderSource{ + started: make(chan struct{}), + release: make(chan struct{}), + } + service, err := NewFreightService( + repository, + source, + fakeClock{}, + &sequenceIDs{}, + time.Minute, + ) + if err != nil { + t.Fatalf("NewFreightService() error = %v", err) + } + firstResult := make(chan error, 1) + go func() { + _, runErr := service.CreateOrderSync( + context.Background(), + validFreightOrderSyncCommand("sync-first"), + ) + firstResult <- runErr + }() + select { + case <-source.started: + case <-time.After(time.Second): + t.Fatal("first order sync did not reach source") + } + startedAt := time.Now() + _, err = service.CreateOrderSync( + context.Background(), + validFreightOrderSyncCommand("sync-second"), + ) + assertUsecaseError(t, err, ErrorKindConflict, "FREIGHT_SYNC_BUSY") + if time.Since(startedAt) > 100*time.Millisecond { + t.Fatalf("busy response took %s", time.Since(startedAt)) + } + close(source.release) + if err := <-firstResult; err != nil { + t.Fatalf("first CreateOrderSync() error = %v", err) + } +} + +func validFreightOrderSyncCommand(idempotencyKey string) CreateFreightSyncCommand { + return CreateFreightSyncCommand{ + CreatorSubject: "local-admin", + ActorUserID: "00000000-0000-4000-8000-000000000099", + IdempotencyKey: idempotencyKey, + OrderNumber: "ORDER-123", + } +} + func TestFreightDateQuerySplitsIntoSevenDayWindows(t *testing.T) { source := &recordingDateSource{} service := &FreightService{source: source} @@ -261,6 +432,144 @@ type recordingDateSource struct { ensureErr error } +type blockingOrderSource struct { + started chan struct{} + release chan struct{} + once sync.Once +} + +func (source *blockingOrderSource) QueryOrder( + ctx context.Context, + _ string, +) (domain.FreightSourceBatch, error) { + if source.started != nil { + source.once.Do(func() { + close(source.started) + }) + } + if source.release != nil { + select { + case <-source.release: + case <-ctx.Done(): + return domain.FreightSourceBatch{}, ctx.Err() + } + } else { + <-ctx.Done() + return domain.FreightSourceBatch{}, ctx.Err() + } + return validFreightSource(), nil +} + +func (*blockingOrderSource) QueryCreatedRange( + context.Context, + string, + string, +) (domain.FreightSourceBatch, error) { + return domain.FreightSourceBatch{}, errors.New("unexpected date query") +} + +type syncTrackingRepository struct { + mu sync.Mutex + startCalls int + completeCalls int + failCalls int + failCode string + failContextErr error +} + +func (*syncTrackingRepository) CreateFreightSync( + _ context.Context, + run domain.FreightSyncRun, + _, _ string, +) (domain.FreightSyncRun, bool, error) { + return run, true, nil +} + +func (repository *syncTrackingRepository) StartFreightSync( + context.Context, + string, + time.Time, +) error { + repository.mu.Lock() + defer repository.mu.Unlock() + repository.startCalls++ + return nil +} + +func (repository *syncTrackingRepository) CompleteFreightSync( + context.Context, + domain.FreightSyncRun, + domain.FreightImportBatch, + time.Time, +) error { + repository.mu.Lock() + defer repository.mu.Unlock() + repository.completeCalls++ + return nil +} + +func (*syncTrackingRepository) CompleteFreightDateSync( + context.Context, + domain.FreightSyncRun, + domain.FreightImportBatch, + time.Time, + time.Time, +) error { + return errors.New("unexpected date completion") +} + +func (repository *syncTrackingRepository) FailFreightSync( + ctx context.Context, + _ string, + code string, + _ time.Time, +) error { + repository.mu.Lock() + defer repository.mu.Unlock() + repository.failCalls++ + repository.failCode = code + repository.failContextErr = ctx.Err() + return nil +} + +func (*syncTrackingRepository) RecoverFreightSyncs( + context.Context, + time.Time, +) (int64, error) { + return 0, nil +} + +func (*syncTrackingRepository) GetFreightSync( + context.Context, + string, + string, +) (domain.FreightSyncRun, error) { + return domain.FreightSyncRun{}, nil +} + +func (*syncTrackingRepository) GetFreightSyncWatermark( + context.Context, + string, +) (*domain.FreightSyncWatermark, error) { + return nil, nil +} + +func (*syncTrackingRepository) ListFreightOrders( + context.Context, + string, + int, +) ([]domain.FreightOrder, error) { + return nil, nil +} + +func (*syncTrackingRepository) GetFreightOrder( + context.Context, + string, + string, +) (domain.FreightOrderDetail, error) { + return domain.FreightOrderDetail{}, nil +} + func (source *recordingDateSource) EnsureAuthenticated(context.Context) error { source.ensureCalls++ return source.ensureErr diff --git a/docs/api.md b/docs/api.md index da57606..f485c23 100644 --- a/docs/api.md +++ b/docs/api.md @@ -242,7 +242,7 @@ ERP 配置来源： ### `POST /api/v1/freight-syncs` -ADMIN 创建异步同步记录，必须带 `Idempotency-Key`： +ADMIN 创建货运同步记录，必须带 `Idempotency-Key`： ```json {"mode":"ORDER_NUMBER","order_number":"完整单号"} @@ -262,8 +262,17 @@ T-224 增加： `{"mode":"CREATED_RANGE","sync_to_now":true}`。后者在没有水位时从当天开始，有水位时 从成功水位前回看 10 分钟对应的自然日开始，后端再按最多 7 天切窗。 -响应 `202`，返回 sync id/status。订单号不进入 URL、事件 message 或访问日志；数据库 -只保存规范值及用于审计/检索的受控字段，不保存 ERP 凭证、Cookie 或 JWT。 +`ORDER_NUMBER` 在同一请求内执行认证、ERP 查询、规范化和事务落库，总预算 55 秒。首次成功 +返回 `201` 和最终 `SUCCEEDED` run，成功幂等重放返回 `200`，两者均设置 +`Location: /api/v1/freight-orders`。同时只能执行一个完整单号同步，其他请求快速返回 +`409 FREIGHT_SYNC_BUSY`；超时返回 `504 FREIGHT_SYNC_TIMEOUT`，并使用独立 cleanup context +把已创建 run 保存为 `FAILED`。Admin Web 成功后跳转 `/freight?notice=import-succeeded`， +此时列表已能读取货运头和全部商品明细。 + +`CREATED_RANGE` 和 `sync_to_now` 保持异步，响应 `202` 和 `PENDING` run，随后在后台执行。 +订单号不进入 URL、事件 message 或访问日志；数据库只保存规范值及用于审计/检索的受控字段， +不保存 ERP 凭证、Cookie 或 JWT。HTTP Server `WriteTimeout` 为 70 秒，ERP 单次请求 timeout +仍为 30 秒。 ### `GET /api/v1/freight-syncs/{sync_id}` @@ -286,8 +295,9 @@ T-224 增加： T-222 返回货运头、全部当前商品明细和 revision/hash 状态。T-223 再增加采购需求和 已生成 task 引用。不存在和跨 creator 统一 404；响应 `Cache-Control: no-store`。 -T-227 的后台 worker 使用当前 Go 内存会话，按 `listTotal -> list 分页 -> listByStock` -查询。每次先校验会话；列表最多 100 条、每页 20 条，详情每批最多 100 个外部 stock ID。 +T-227 的 Go source 使用当前内存会话，按 `listTotal -> list 分页 -> listByStock` 查询； +完整单号由请求同步调用，日期范围由后台 worker 调用。每次先校验会话；列表最多 100 条、 +每页 20 条，详情每批最多 100 个外部 stock ID。 未配置、未登录、找不到货运单、响应协议错误和暂时不可用分别落为 `ERP_NOT_CONFIGURED`、`ERP_SESSION_REQUIRED`、`ERP_FREIGHT_NOT_FOUND`、 `ERP_RESPONSE_INVALID` 和 `ERP_UNAVAILABLE`，不返回 ERP 原始错误 body。 diff --git a/docs/current-state.md b/docs/current-state.md index 517b1d8..07f4b92 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -5,7 +5,7 @@ ## 当前快照 - 日期：2026-07-29 -- 阶段：T-237 已规划完整单号 55 秒同步导入，待实现 +- 阶段：T-237 已完成完整单号 55 秒同步导入 - Git：当前分支为 `main`；T-001 至 T-004、T-101 至 T-104、T-201 至 T-219 均按文档提交、实现提交的顺序纳入历史 - 生产代码：`android-buyer/` 已接入 Roubao Android 源码 @@ -16,8 +16,9 @@ task-scoped 参考图和 `authctl` - ERP 货运：Go 后端以受锁内存 ERP 会话直连按完整单号或创建日期同步；v12 保存 同步记录、货运头和全部明细，canonical hash 控制 revision，Admin 已有 `/freight`、 - `/freight/import`、`/freight/{id}` 与对应 JSON API。T-237 计划将完整单号改为 55 秒 - 同步响应并可靠写终态，日期范围仍后台异步。 + `/freight/import`、`/freight/{id}` 与对应 JSON API。T-237 已将完整单号改为 55 秒 + 同步响应，成功后直接进入可见货运列表；超时/取消使用独立 cleanup context 写 + `FAILED`，日期范围仍后台异步。HTTP `WriteTimeout` 为 70 秒。 - ERP Go 迁移：T-225 已用脱敏 fixture 固定 `internal/platform/shunyunbao` 的 header、 单号/日期查询、分页、详情批量和字段 allowlist，并使货运用例依赖来源中立错误。T-226 已增加受锁保护的 Go 内存 Cookie jar、验证码 ticket、登录和用户校验，以及 ADMIN 的 @@ -39,7 +40,7 @@ - Android Studio：未安装；`winget` 静默安装卡住后已终止，不阻塞命令行构建 - 测试：T-219 Android Debug/Release 单元测试与构建和根 `init.ps1` 通过； Debug APK `1.4.16 (21)` 已覆盖安装到 PKG110 -- 后端测试：T-226 至 T-229 已运行 `go test ./...`、`go test -race ./...`、`go vet ./...` +- 后端测试：T-226 至 T-237 已运行 `go test ./...`、`go test -race ./...`、`go vet ./...` 和三个 Go 入口构建；T-227 增加 Go source 的伪 ERP 会话预检、完整单号、日期分页去重、 详情 allowlist 和稳定错误码覆盖；根 `init.ps1` 的 Android 测试/Debug APK 与 Go 标准 验证也通过，均未访问真实 ERP； diff --git a/docs/tasks/T-237.md b/docs/tasks/T-237.md index 18668eb..63a6b70 100644 --- a/docs/tasks/T-237.md +++ b/docs/tasks/T-237.md @@ -4,7 +4,7 @@ title: 完整单号货运同步导入 phase: 2 deps: - T-236 -status: PLANNED +status: DONE created: 2026-07-29 context_ref: cf83fb5 work_branch: null @@ -45,14 +45,14 @@ Admin 按完整订单号导入 ERP 货运时，当前请求只完成登录预检 ## 验收要点 -- [ ] 完整单号 POST 只在货运头和明细事务提交后返回成功，返回时 run 为 `SUCCEEDED`。 -- [ ] Admin 成功后跳转 `/freight`，列表立即显示本次货运单；页面不再要求刷新 `RUNNING`。 -- [ ] 完整单号从请求进入到执行结束最多 55 秒；超时可靠保存 `FAILED/FREIGHT_SYNC_TIMEOUT`。 -- [ ] ERP、协议、存储和请求取消也可靠保存终态，不复用已取消 context 进行失败收尾。 -- [ ] 同进程并发完整单号导入快速返回 `FREIGHT_SYNC_BUSY`；幂等成功重放不重复执行。 -- [ ] 日期范围导入继续返回 `202/PENDING` 并在后台执行。 -- [ ] HTTP `WriteTimeout` 为 70 秒，覆盖同步业务预算。 -- [ ] 标准 Go 测试、race、vet 和三个入口构建通过。 +- [x] 完整单号 POST 只在货运头和明细事务提交后返回成功，返回时 run 为 `SUCCEEDED`。 +- [x] Admin 成功后跳转 `/freight`，列表立即显示本次货运单；页面不再要求刷新 `RUNNING`。 +- [x] 完整单号从请求进入到执行结束最多 55 秒；超时可靠保存 `FAILED/FREIGHT_SYNC_TIMEOUT`。 +- [x] ERP、协议、存储和请求取消也可靠保存终态，不复用已取消 context 进行失败收尾。 +- [x] 同进程并发完整单号导入快速返回 `FREIGHT_SYNC_BUSY`；幂等成功重放不重复执行。 +- [x] 日期范围导入继续返回 `202/PENDING` 并在后台执行。 +- [x] HTTP `WriteTimeout` 为 70 秒，覆盖同步业务预算。 +- [x] 标准 Go 测试、race、vet 和三个入口构建通过。 ## 边界 @@ -64,3 +64,7 @@ Admin 按完整订单号导入 ERP 货运时，当前请求只完成登录预检 - 2026-07-29：创建任务。确认同步化范围仅为完整单号；55 秒是整个单号请求预算，服务端写 超时提高至 70 秒。终态收尾、并发拒绝和幂等重放与同步响应一起实现。 +- 2026-07-29：完整单号改为请求内同步执行，成功 API 返回 `201/SUCCEEDED`、重放返回 + `200`，Admin 跳转可立即读取的货运列表；日期同步保留 `202` 后台模式。增加 55 秒预算、 + 70 秒 WriteTimeout、单进程并发门、超时/取消独立失败收尾，以及 success/timeout/busy/ + session/not-found 的稳定 Web/API 映射。标准 Go 测试、race、vet 及三个入口构建均通过。