package httpapi import ( "bytes" "context" "crypto/sha256" "database/sql" "encoding/hex" "errors" "fmt" "image" "image/color" "image/jpeg" "io" "net/http" "net/http/httptest" "path/filepath" "strconv" "strings" "sync" "testing" "time" "cmroubao/backend-api/internal/domain" "cmroubao/backend-api/internal/platform/assetstore" "cmroubao/backend-api/internal/platform/database" "cmroubao/backend-api/internal/platform/migration" repository "cmroubao/backend-api/internal/repository/sqlite" "cmroubao/backend-api/internal/usecase" "github.com/gin-gonic/gin" ) const ( deviceTestAdminID = "00000000-0000-4000-8000-000000000201" deviceTestBuyerID = "00000000-0000-4000-8000-000000000202" deviceTestDeviceID = "00000000-0000-4000-8000-000000000203" ) func TestDeviceRoutesRequireBuyerBearerAndSeparateRoles(t *testing.T) { fixture := newDeviceHTTPFixture(t) body := `{ "app_version":"0.1.0", "android_version":"16", "pdd_version":"8.17.0", "readiness":{ "accessibility_enabled":true, "pdd_installed":true, "active_task_id":null } }` withoutBearer := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/devices/heartbeat", contentType: "application/json", body: strings.NewReader(body), }) if withoutBearer.Code != http.StatusUnauthorized { t.Fatalf( "request without bearer status/body = %d / %s", withoutBearer.Code, withoutBearer.Body, ) } assertErrorCode(t, withoutBearer, "DEVICE_ACCESS_REQUIRED") adminBearer := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/devices/heartbeat", contentType: "application/json", body: strings.NewReader(body), bearerToken: testCSRFOpaqueToken, }) if adminBearer.Code != http.StatusUnauthorized { t.Fatalf( "admin bearer status/body = %d / %s", adminBearer.Code, adminBearer.Body, ) } assertErrorCode(t, adminBearer, "DEVICE_ACCESS_REQUIRED") adminCookieRequest := httptest.NewRequest( http.MethodPost, "/api/v1/devices/heartbeat", strings.NewReader(body), ) adminCookieRequest.Header.Set("Content-Type", "application/json") adminCookieRequest.AddCookie(&http.Cookie{ Name: "cmroubao_admin_session", Value: testOpaqueToken, }) adminCookieResponse := httptest.NewRecorder() fixture.router.ServeHTTP(adminCookieResponse, adminCookieRequest) if adminCookieResponse.Code != http.StatusUnauthorized { t.Fatalf( "admin cookie on device route status/body = %d / %s", adminCookieResponse.Code, adminCookieResponse.Body, ) } assertErrorCode( t, adminCookieResponse, "DEVICE_ACCESS_REQUIRED", ) buyerOnAdmin := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodGet, target: "/api/v1/admin-probe", bearerToken: testOpaqueToken, }) if buyerOnAdmin.Code != http.StatusUnauthorized { t.Fatalf( "buyer bearer on admin route status/body = %d / %s", buyerOnAdmin.Code, buyerOnAdmin.Body, ) } assertErrorCode(t, buyerOnAdmin, "ADMIN_SESSION_REQUIRED") } func TestDeviceHeartbeatUsesPrincipalDeviceAndRejectsBodyMismatch( t *testing.T, ) { fixture := newDeviceHTTPFixture(t) mismatch := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/devices/heartbeat", contentType: "application/json", body: strings.NewReader(`{ "device_id":"00000000-0000-4000-8000-000000000299", "app_version":"0.1.0", "android_version":"16", "pdd_version":"8.17.0", "readiness":{ "accessibility_enabled":true, "pdd_installed":true } }`), bearerToken: testOpaqueToken, }) if mismatch.Code != http.StatusForbidden { t.Fatalf( "device mismatch status/body = %d / %s", mismatch.Code, mismatch.Body, ) } assertErrorCode(t, mismatch, "DEVICE_ID_MISMATCH") heartbeat := fixture.readyHeartbeat(t) if heartbeat.Code != http.StatusOK { t.Fatalf( "heartbeat status/body = %d / %s", heartbeat.Code, heartbeat.Body, ) } var response struct { DeviceID string `json:"device_id"` Readiness struct { AccessibilityEnabled bool `json:"accessibility_enabled"` PDDInstalled bool `json:"pdd_installed"` } `json:"readiness"` ClientStateMatches bool `json:"client_state_matches"` } decodeResponse(t, heartbeat, &response) if response.DeviceID != deviceTestDeviceID || !response.Readiness.AccessibilityEnabled || !response.Readiness.PDDInstalled || !response.ClientStateMatches { t.Fatalf("heartbeat response = %+v", response) } if heartbeat.Header().Get("Cache-Control") != "no-store" { t.Fatalf( "heartbeat Cache-Control = %q", heartbeat.Header().Get("Cache-Control"), ) } } func TestDeviceRoutesRejectUnsupportedAndUnknownJSON(t *testing.T) { fixture := newDeviceHTTPFixture(t) body := `{ "app_version":"0.1.0", "android_version":"16", "pdd_version":"8.17.0", "readiness":{ "accessibility_enabled":true, "pdd_installed":true } }` unsupported := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/devices/heartbeat", body: strings.NewReader(body), bearerToken: testOpaqueToken, }) if unsupported.Code != http.StatusUnsupportedMediaType { t.Fatalf( "unsupported media status/body = %d / %s", unsupported.Code, unsupported.Body, ) } assertErrorCode(t, unsupported, "UNSUPPORTED_MEDIA_TYPE") unknownField := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/devices/heartbeat", contentType: "application/json", body: strings.NewReader(`{ "app_version":"0.1.0", "android_version":"16", "pdd_version":"8.17.0", "readiness":{ "accessibility_enabled":true, "pdd_installed":true }, "unexpected":true }`), bearerToken: testOpaqueToken, }) if unknownField.Code != http.StatusBadRequest { t.Fatalf( "unknown JSON status/body = %d / %s", unknownField.Code, unknownField.Body, ) } assertErrorCode(t, unknownField, "INVALID_JSON") } func TestDeviceClaimReturnsTaskOrNoContentWithoutClaimSecret( t *testing.T, ) { fixture := newDeviceHTTPFixture(t) requireDeviceStatus(t, fixture.readyHeartbeat(t), http.StatusOK) noTask := fixture.claimNext(t, "claim-empty", testOpaqueToken) requireDeviceStatus(t, noTask, http.StatusNoContent) assertNoClaimSecret(t, noTask, testOpaqueToken) taskID := fixture.createPendingTask(t) claimed := fixture.claimNext(t, "claim-task", testOpaqueToken) requireDeviceStatus(t, claimed, http.StatusOK) var response deviceLifecycleResponse decodeResponse(t, claimed, &response) if response.Task.ID != taskID || response.Task.Status != string(domain.TaskStatusClaimed) || response.Task.Version < 2 || response.Task.ClaimGeneration != 1 || response.Task.ReferenceImageURL == "" || response.Replayed { t.Fatalf("claim response = %+v", response) } assertNoClaimSecret(t, claimed, testOpaqueToken) wrongToken := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodGet, target: response.Task.ReferenceImageURL, bearerToken: testOpaqueToken, claimToken: testCSRFOpaqueToken, }) requireDeviceStatus(t, wrongToken, http.StatusForbidden) assertErrorCode(t, wrongToken, "TASK_CLAIM_INVALID") imageResponse := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodGet, target: response.Task.ReferenceImageURL, bearerToken: testOpaqueToken, claimToken: testOpaqueToken, }) requireDeviceStatus(t, imageResponse, http.StatusOK) if imageResponse.Header().Get("Content-Type") != "image/jpeg" || imageResponse.Header().Get("Cache-Control") != "private, no-store" || imageResponse.Body.Len() == 0 { t.Fatalf( "reference image headers/body = %#v / %d", imageResponse.Header(), imageResponse.Body.Len(), ) } assertNoClaimSecret(t, imageResponse, testOpaqueToken) } func TestDeviceConcurrentHTTPClaimKeepsOneActiveTaskPerDevice( t *testing.T, ) { fixture := newDeviceHTTPFixture(t) requireDeviceStatus(t, fixture.readyHeartbeat(t), http.StatusOK) fixture.createPendingTask(t) fixture.createPendingTask(t) server := httptest.NewServer(fixture.router) t.Cleanup(server.Close) type outcome struct { status int body string err error } outcomes := make(chan outcome, 2) start := make(chan struct{}) var workers sync.WaitGroup for index, token := range []string{ testOpaqueToken, testCSRFOpaqueToken, } { index := index token := token workers.Add(1) go func() { defer workers.Done() <-start request, err := http.NewRequest( http.MethodPost, server.URL+"/api/v1/tasks/claim-next", strings.NewReader(`{}`), ) if err != nil { outcomes <- outcome{err: err} return } request.Header.Set("Content-Type", "application/json") request.Header.Set("Authorization", "Bearer "+testOpaqueToken) request.Header.Set(claimTokenHeader, token) request.Header.Set( "Idempotency-Key", fmt.Sprintf("concurrent-http-claim-%d", index), ) response, err := server.Client().Do(request) if err != nil { outcomes <- outcome{err: err} return } defer response.Body.Close() body, err := io.ReadAll(response.Body) outcomes <- outcome{ status: response.StatusCode, body: string(body), err: err, } }() } close(start) workers.Wait() close(outcomes) statuses := map[int]int{} for result := range outcomes { if result.err != nil { t.Fatalf("concurrent HTTP claim error = %v", result.err) } statuses[result.status]++ if strings.Contains( strings.ToLower(result.body), "claim_token", ) { t.Fatalf("concurrent response leaked claim field: %s", result.body) } } if statuses[http.StatusOK] != 1 || statuses[http.StatusConflict] != 1 { t.Fatalf("concurrent HTTP statuses = %#v", statuses) } var active int if err := fixture.db.QueryRow( `SELECT COUNT(*) FROM purchase_tasks WHERE claimed_by_device_id = ? AND status = 'CLAIMED'`, deviceTestDeviceID, ).Scan(&active); err != nil { t.Fatalf("count active HTTP claims: %v", err) } if active != 1 { t.Fatalf("active HTTP claims = %d", active) } } func TestDeviceStartAndTaskHeartbeatUseClaimContract(t *testing.T) { fixture := newDeviceHTTPFixture(t) requireDeviceStatus(t, fixture.readyHeartbeat(t), http.StatusOK) taskID := fixture.createPendingTask(t) claim := fixture.claimNext(t, "claim-for-start", testOpaqueToken) requireDeviceStatus(t, claim, http.StatusOK) var claimed deviceLifecycleResponse decodeResponse(t, claim, &claimed) startBody := fmt.Sprintf( `{"claim_generation":%d,"expected_version":%d}`, claimed.Task.ClaimGeneration, claimed.Task.Version, ) wrongToken := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/start", contentType: "application/json", body: strings.NewReader(startBody), bearerToken: testOpaqueToken, claimToken: testCSRFOpaqueToken, idempotencyKey: "start-wrong-token", }) if wrongToken.Code != http.StatusForbidden { t.Fatalf( "wrong claim token status/body = %d / %s", wrongToken.Code, wrongToken.Body, ) } assertErrorCode(t, wrongToken, "TASK_CLAIM_INVALID") startedResponse := performDeviceRequest( t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/start", contentType: "application/json", body: strings.NewReader(startBody), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "start-correct-token", }, ) requireDeviceStatus(t, startedResponse, http.StatusOK) var started deviceLifecycleResponse decodeResponse(t, startedResponse, &started) if started.Task.Status != string(domain.TaskStatusRunning) || started.Execution.ID == "" || started.Execution.CurrentStep != "PREFLIGHT" || started.Execution.ExpiresAt.IsZero() || started.Execution.OrderSubmitted { t.Fatalf("start response = %+v", started) } if remaining := time.Until(started.Execution.ExpiresAt); remaining < 9*time.Minute || remaining > 11*time.Minute { t.Fatalf("start execution expiry remaining = %s", remaining) } assertNoClaimSecret(t, startedResponse, testOpaqueToken) heartbeatBody := fmt.Sprintf( `{ "execution_id":%q, "claim_generation":%d, "step":"SEARCH_RESULTS" }`, started.Execution.ID, started.Task.ClaimGeneration, ) taskHeartbeat := performDeviceRequest( t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/heartbeat", contentType: "application/json", body: strings.NewReader(heartbeatBody), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, }, ) requireDeviceStatus(t, taskHeartbeat, http.StatusOK) var heartbeat deviceLifecycleResponse decodeResponse(t, taskHeartbeat, &heartbeat) if heartbeat.Task.Status != string(domain.TaskStatusRunning) || heartbeat.Execution.ID != started.Execution.ID || heartbeat.Execution.CurrentStep != "SEARCH_RESULTS" || heartbeat.Execution.ExpiresAt.IsZero() || heartbeat.Execution.ExpiresAt.Before(started.Execution.ExpiresAt) || heartbeat.Execution.OrderSubmitted || heartbeat.CancelRequested { t.Fatalf("task heartbeat response = %+v", heartbeat) } assertNoClaimSecret(t, taskHeartbeat, testOpaqueToken) } func TestDeviceExecutionResultsAreIdempotentAndAuditable(t *testing.T) { fixture := newDeviceHTTPFixture(t) requireDeviceStatus(t, fixture.readyHeartbeat(t), http.StatusOK) taskID := fixture.createPendingTask(t) claimedResponse := fixture.claimNext(t, "claim-for-results", testOpaqueToken) requireDeviceStatus(t, claimedResponse, http.StatusOK) var claimed deviceLifecycleResponse decodeResponse(t, claimedResponse, &claimed) startResponse := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/start", contentType: "application/json", body: strings.NewReader(fmt.Sprintf( `{"claim_generation":%d,"expected_version":%d}`, claimed.Task.ClaimGeneration, claimed.Task.Version, )), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "start-for-results", }) requireDeviceStatus(t, startResponse, http.StatusOK) var started deviceLifecycleResponse decodeResponse(t, startResponse, &started) occurredAt := time.Now().UTC().Format(time.RFC3339Nano) events := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/events", contentType: "application/json", body: strings.NewReader(fmt.Sprintf( `{"execution_id":%q,"claim_generation":%d,"events":[{"event_id":"00000000-0000-4000-8000-000000000701","step":"SEARCH","type":"SEARCH_STARTED","message":"开始采集候选","occurred_at":%q}]}`, started.Execution.ID, started.Task.ClaimGeneration, occurredAt, )), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "result-events-1", }) requireDeviceStatus(t, events, http.StatusOK) detailEvidence := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/evidence", contentType: "image/jpeg", body: deviceReferenceImage(t, 701), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "result-evidence-1", executionID: started.Execution.ID, claimGeneration: started.Task.ClaimGeneration, }) requireDeviceStatus(t, detailEvidence, http.StatusCreated) var evidenceResponse struct { Evidence struct { ID string `json:"id"` SHA256 string `json:"sha256"` } `json:"evidence"` } decodeResponse(t, detailEvidence, &evidenceResponse) if evidenceResponse.Evidence.ID == "" || evidenceResponse.Evidence.SHA256 == "" { t.Fatalf("detail evidence response = %s", detailEvidence.Body.String()) } specificationEvidence := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/evidence", contentType: "image/jpeg", body: deviceReferenceImage(t, 702), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "result-evidence-2", executionID: started.Execution.ID, claimGeneration: started.Task.ClaimGeneration, }) requireDeviceStatus(t, specificationEvidence, http.StatusCreated) var specificationEvidenceResponse struct { Evidence struct { ID string `json:"id"` SHA256 string `json:"sha256"` } `json:"evidence"` } decodeResponse(t, specificationEvidence, &specificationEvidenceResponse) if specificationEvidenceResponse.Evidence.ID == "" || specificationEvidenceResponse.Evidence.SHA256 == "" { t.Fatalf( "specification evidence response = %s", specificationEvidence.Body.String(), ) } detail, err := fixture.tasks.Get(context.Background(), "local-admin", taskID) if err != nil { t.Fatalf("get task for content hash: %v", err) } taskHash := usecase.TaskContentSHA256(detail.Task) cardSignature := strings.Repeat("b", 64) detailSignature := strings.Repeat("c", 64) candidatePayload := fmt.Sprintf( `{"execution_id":%q,"claim_generation":%d,"task_content_sha256":%q,"execution_mode":"MANUAL_FIRST","search_query":"TEST-SKU","candidates":[{"ordinal":1,"title":"手动候选","sku_text":"TEST-SKU","price":"12.00","product_url":"","image_url":"","card_signature":%q,"detail_signature":%q,"detail_evidence_sha256":%q,"specification_evidence_sha256":%q,"evidence_asset_ids":[%q,%q],"evaluation":null}]}`, started.Execution.ID, started.Task.ClaimGeneration, taskHash, cardSignature, detailSignature, evidenceResponse.Evidence.SHA256, specificationEvidenceResponse.Evidence.SHA256, evidenceResponse.Evidence.ID, specificationEvidenceResponse.Evidence.ID, ) badCandidatePayload := strings.Replace( candidatePayload, evidenceResponse.Evidence.SHA256, strings.Repeat("0", 64), 1, ) badCandidates := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/candidates", contentType: "application/json", body: strings.NewReader(badCandidatePayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "result-candidates-bad-hash", }) requireDeviceStatus(t, badCandidates, http.StatusConflict) candidates := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/candidates", contentType: "application/json", body: strings.NewReader(candidatePayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "result-candidates-1", }) requireDeviceStatus(t, candidates, http.StatusOK) candidateReplay := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/candidates", contentType: "application/json", body: strings.NewReader(candidatePayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "result-candidates-1", }) requireDeviceStatus(t, candidateReplay, http.StatusOK) if !strings.Contains(candidateReplay.Body.String(), `"replayed":true`) { t.Fatalf("candidate replay response = %s", candidateReplay.Body.String()) } detail, err = fixture.tasks.Get(context.Background(), "local-admin", taskID) if err != nil { t.Fatalf("get task after candidate upload: %v", err) } readyEvents := 0 for _, event := range detail.Events { if event.Type == "CANDIDATES_READY" { readyEvents++ } } if detail.Task.Status != domain.TaskStatusWaitingConfirmation || readyEvents != 1 { t.Fatalf( "candidate upload status/events = %s/%+v", detail.Task.Status, detail.Events, ) } humanReviewPayload := fmt.Sprintf( `{"execution_id":%q,"claim_generation":%d,"task_content_sha256":%q,"reason_schema_version":1,"outcome":"CANDIDATE_ACCEPTED","selected_candidate_ordinal":1,"primary_reason_code":"SELECTED_BEST_MATCH","note":"","items":[{"candidate_ordinal":1,"label":"ACCEPT","primary_reason_code":"SKU_MATCH","reason_codes":["SKU_MATCH"],"note":""}]}`, started.Execution.ID, started.Task.ClaimGeneration, taskHash, ) humanReview := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/human-reviews", contentType: "application/json", body: strings.NewReader(humanReviewPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "result-human-review-1", }) requireDeviceStatus(t, humanReview, http.StatusOK) if !strings.Contains(humanReview.Body.String(), `"version":1`) { t.Fatalf("human review response = %s", humanReview.Body.String()) } var storedReview struct { Review struct { ID string `json:"id"` } `json:"review"` } decodeResponse(t, humanReview, &storedReview) if storedReview.Review.ID == "" { t.Fatalf("human review ID missing: %+v", storedReview) } humanReviewReplay := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/human-reviews", contentType: "application/json", body: strings.NewReader(humanReviewPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "result-human-review-1", }) requireDeviceStatus(t, humanReviewReplay, http.StatusOK) if !strings.Contains(humanReviewReplay.Body.String(), `"replayed":true`) { t.Fatalf("human review replay response = %s", humanReviewReplay.Body.String()) } revisedReviewPayload := fmt.Sprintf( `{"execution_id":%q,"claim_generation":%d,"task_content_sha256":%q,"reason_schema_version":1,"outcome":"CANDIDATE_ACCEPTED","selected_candidate_ordinal":1,"primary_reason_code":"SELECTED_BEST_MATCH","note":"","supersedes_review_id":%q,"items":[{"candidate_ordinal":1,"label":"ACCEPT","primary_reason_code":"IMAGE_MATCH","reason_codes":["IMAGE_MATCH"],"note":""}]}`, started.Execution.ID, started.Task.ClaimGeneration, taskHash, storedReview.Review.ID, ) revisedReview := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/human-reviews", contentType: "application/json", body: strings.NewReader(revisedReviewPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "result-human-review-2", }) requireDeviceStatus(t, revisedReview, http.StatusOK) if !strings.Contains(revisedReview.Body.String(), `"version":2`) { t.Fatalf("revised human review response = %s", revisedReview.Body.String()) } runner, err := migration.New(fixture.db) if err != nil { t.Fatalf("migration.New() after review error = %v", err) } if err := runner.Down(context.Background()); err != nil { t.Fatalf("date sync migration down: %v", err) } if err := runner.Down(context.Background()); err != nil { t.Fatalf("procurement migration down: %v", err) } if err := runner.Down(context.Background()); err != nil { t.Fatalf("freight migration down: %v", err) } if err := runner.Down(context.Background()); err != nil { t.Fatalf("order submission migration down: %v", err) } if err := runner.Down(context.Background()); err != nil { t.Fatalf("order dry-run migration down: %v", err) } if err := runner.Down(context.Background()); err != nil { t.Fatalf("device command migration down: %v", err) } if err := runner.Down(context.Background()); err == nil { t.Fatal("order workflow migration down succeeded with retained data") } if applied, err := runner.Up(context.Background()); err != nil { t.Fatalf("restore device command migration: %v", err) } else if applied != 6 { t.Fatalf("restored migrations = %d, want 6", applied) } completePayload := fmt.Sprintf( `{"execution_id":%q,"claim_generation":%d,"task_content_sha256":%q,"execution_mode":"MANUAL_FIRST","outcome":"CANDIDATE_ACCEPTED","operator_reason":"人工核对标题、SKU和截图后接受","candidate":{"ordinal":1,"title":"手动候选","sku_text":"TEST-SKU","price":"12.00","product_url":"","image_url":"","card_signature":%q,"detail_signature":%q,"detail_evidence_sha256":%q,"specification_evidence_sha256":%q,"evidence_asset_ids":[%q,%q],"evaluation":null},"order_submitted":false}`, started.Execution.ID, started.Task.ClaimGeneration, taskHash, cardSignature, detailSignature, evidenceResponse.Evidence.SHA256, specificationEvidenceResponse.Evidence.SHA256, evidenceResponse.Evidence.ID, specificationEvidenceResponse.Evidence.ID, ) complete := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/complete", contentType: "application/json", body: strings.NewReader(completePayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "result-complete-1", }) requireDeviceStatus(t, complete, http.StatusOK) assertNoClaimSecret(t, complete, testOpaqueToken) replayed := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/complete", contentType: "application/json", body: strings.NewReader(completePayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "result-complete-1", }) requireDeviceStatus(t, replayed, http.StatusOK) if !strings.Contains(replayed.Body.String(), `"replayed":true`) { t.Fatalf("terminal replay response = %s", replayed.Body.String()) } detail, err = fixture.tasks.Get(context.Background(), "local-admin", taskID) if err != nil { t.Fatalf("get task result detail: %v", err) } if detail.Task.Status != domain.TaskStatusSucceeded || detail.Report == nil || detail.Report.Outcome == nil || detail.Report.Outcome.OrderSubmitted || len(detail.Report.Events) != 1 || len(detail.Report.EvidenceAssets) != 2 || detail.Report.CandidateBatch == nil || detail.Report.DecisionDataset == nil || len(detail.Report.DecisionDataset.Observations) != 1 || len(detail.Report.DecisionDataset.HumanReviews) != 2 || detail.Report.DecisionDataset.HumanReviews[0].Version != 1 || detail.Report.DecisionDataset.HumanReviews[1].Version != 2 { t.Fatalf("execution report = %+v", detail.Report) } identity := detail.Report.DecisionDataset.Observations[0].Identity if identity == nil || len(identity.CandidateKey) != 64 || identity.CardSignature != cardSignature || identity.DetailSignature != detailSignature || identity.DetailEvidenceSHA256 != evidenceResponse.Evidence.SHA256 || identity.SpecificationEvidenceSHA256 != specificationEvidenceResponse.Evidence.SHA256 { t.Fatalf("candidate identity = %+v", identity) } } func TestDeviceOrderCommandDeliveryAndAcknowledgementAreRecoverable( t *testing.T, ) { fixture := newDeviceHTTPFixture(t) requireDeviceStatus(t, fixture.readyHeartbeat(t), http.StatusOK) taskID := fixture.createPendingTask(t) claim := fixture.claimNext(t, "order-command-claim", testOpaqueToken) requireDeviceStatus(t, claim, http.StatusOK) var claimed deviceLifecycleResponse decodeResponse(t, claim, &claimed) startPayload := fmt.Sprintf( `{"claim_generation":%d,"expected_version":%d}`, claimed.Task.ClaimGeneration, claimed.Task.Version, ) start := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/start", contentType: "application/json", body: strings.NewReader(startPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-command-start", }) requireDeviceStatus(t, start, http.StatusOK) var started deviceLifecycleResponse decodeResponse(t, start, &started) candidateKey := seedDeviceOrderCommandCandidate( t, fixture, taskID, started.Execution.ID, ) detail, err := fixture.tasks.Get( context.Background(), localAdminSubject, taskID, ) if err != nil { t.Fatalf("get waiting task detail: %v", err) } authorizations, err := usecase.NewOrderAuthorizationService( fixture.store, usecase.SystemClock{}, usecase.UUIDGenerator{}, ) if err != nil { t.Fatalf("NewOrderAuthorizationService() error = %v", err) } created, err := authorizations.Create( context.Background(), usecase.CreateOrderAuthorizationCommand{ ActorUserID: deviceTestAdminID, TaskID: taskID, IdempotencyKey: "order-command-authorization", ExecutionID: started.Execution.ID, TaskContentSHA256: usecase.TaskContentSHA256(detail.Task), ExpectedTaskVersion: detail.Task.Version, CandidateKey: candidateKey, ReasonSchemaVersion: 1, PrimaryReasonCode: "SELECTED_BEST_MATCH", Items: []usecase.OrderAuthorizationItemInput{{ CandidateKey: candidateKey, Label: "ACCEPT", PrimaryReasonCode: "SKU_MATCH", ReasonCodes: []string{"SKU_MATCH", "IMAGE_MATCH"}, }}, }, ) if err != nil { t.Fatalf("create order authorization: %v", err) } pullPayload := fmt.Sprintf( `{"device_id":%q,"execution_id":%q,"claim_generation":%d}`, deviceTestDeviceID, started.Execution.ID, started.Task.ClaimGeneration, ) pull := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/commands/next", contentType: "application/json", body: strings.NewReader(pullPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, }) requireDeviceStatus(t, pull, http.StatusOK) var command struct { ID string `json:"id"` Type string `json:"type"` SchemaVersion int `json:"schema_version"` TaskID string `json:"task_id"` ExecutionID string `json:"execution_id"` OriginalSKU string `json:"original_sku"` Quantity int `json:"quantity"` CommandSHA256 string `json:"command_sha256"` AuthorizationStatus string `json:"authorization_status"` Candidate struct { Key string `json:"candidate_key"` ObservedOrdinal int `json:"observed_ordinal"` Title string `json:"title"` SKUText string `json:"sku_text"` CardSignature string `json:"card_signature"` DetailSignature string `json:"detail_signature"` DetailEvidence string `json:"detail_evidence_sha256"` SpecificationSHA string `json:"specification_evidence_sha256"` } `json:"candidate"` } decodeResponse(t, pull, &command) if command.ID != created.Authorization.ID || command.Type != "CREATE_PENDING_ORDER" || command.SchemaVersion != 1 || command.TaskID != taskID || command.ExecutionID != started.Execution.ID || command.OriginalSKU == "" || command.Quantity != 2 || len(command.CommandSHA256) != 64 || command.AuthorizationStatus != "DELIVERED" || command.Candidate.Key != candidateKey || command.Candidate.ObservedOrdinal != 1 || command.Candidate.Title != "设备命令候选" || command.Candidate.SKUText != "TEST-SKU-COMMAND" { t.Fatalf("order command = %+v", command) } replayedPull := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/commands/next", contentType: "application/json", body: strings.NewReader(pullPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, }) requireDeviceStatus(t, replayedPull, http.StatusOK) if !strings.Contains( replayedPull.Body.String(), `"command_sha256":"`+command.CommandSHA256+`"`, ) { t.Fatalf("replayed command = %s", replayedPull.Body.String()) } badAckPayload := fmt.Sprintf( `{"device_id":%q,"execution_id":%q,"claim_generation":%d,"command_sha256":%q}`, deviceTestDeviceID, started.Execution.ID, started.Task.ClaimGeneration, strings.Repeat("0", 64), ) badAck := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/commands/" + command.ID + "/ack", contentType: "application/json", body: strings.NewReader(badAckPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-command-ack-bad", }) requireDeviceStatus(t, badAck, http.StatusConflict) ackPayload := strings.Replace( badAckPayload, strings.Repeat("0", 64), command.CommandSHA256, 1, ) ack := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/commands/" + command.ID + "/ack", contentType: "application/json", body: strings.NewReader(ackPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-command-ack", }) requireDeviceStatus(t, ack, http.StatusOK) if !strings.Contains(ack.Body.String(), `"status":"ACKNOWLEDGED"`) { t.Fatalf("ack response = %s", ack.Body.String()) } ackReplay := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/commands/" + command.ID + "/ack", contentType: "application/json", body: strings.NewReader(ackPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-command-ack", }) requireDeviceStatus(t, ackReplay, http.StatusOK) if !strings.Contains(ackReplay.Body.String(), `"replayed":true`) { t.Fatalf("ack replay = %s", ackReplay.Body.String()) } acknowledgedPull := performDeviceRequest( t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/commands/next", contentType: "application/json", body: strings.NewReader(pullPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, }, ) requireDeviceStatus(t, acknowledgedPull, http.StatusOK) if !strings.Contains( acknowledgedPull.Body.String(), `"authorization_status":"ACKNOWLEDGED"`, ) { t.Fatalf("acknowledged pull = %s", acknowledgedPull.Body.String()) } startDryRunPayload := fmt.Sprintf( `{"device_id":%q,"execution_id":%q,"claim_generation":%d,"command_id":%q,"command_sha256":%q}`, deviceTestDeviceID, started.Execution.ID, started.Task.ClaimGeneration, command.ID, command.CommandSHA256, ) startDryRun := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-dry-runs/start", contentType: "application/json", body: strings.NewReader(startDryRunPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-dry-run-start", }) requireDeviceStatus(t, startDryRun, http.StatusOK) if !strings.Contains(startDryRun.Body.String(), `"status":"PREPARING"`) { t.Fatalf("dry-run start = %s", startDryRun.Body.String()) } startDryRunReplay := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-dry-runs/start", contentType: "application/json", body: strings.NewReader(startDryRunPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-dry-run-start", }) requireDeviceStatus(t, startDryRunReplay, http.StatusOK) if !strings.Contains(startDryRunReplay.Body.String(), `"replayed":true`) { t.Fatalf("dry-run start replay = %s", startDryRunReplay.Body.String()) } executingPull := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/commands/next", contentType: "application/json", body: strings.NewReader(pullPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, }) requireDeviceStatus(t, executingPull, http.StatusOK) if !strings.Contains( executingPull.Body.String(), `"authorization_status":"EXECUTING"`, ) { t.Fatalf("executing pull = %s", executingPull.Body.String()) } const dryRunEvidenceID = "00000000-0000-4000-8000-000000000077" dryRunEvidenceSHA := strings.Repeat("e", 64) if _, err := fixture.db.Exec( `INSERT INTO execution_evidence_assets ( id, task_id, execution_id, media_type, size_bytes, sha256, storage_key, created_at, received_after_execution_expiry ) VALUES (?, ?, ?, 'image/jpeg', 10, ?, ?, ?, 0)`, dryRunEvidenceID, taskID, started.Execution.ID, dryRunEvidenceSHA, "dry-run/order-confirmation.jpg", time.Now().UTC().Format(time.RFC3339Nano), ); err != nil { t.Fatalf("seed dry-run evidence: %v", err) } readyDryRunPayload := fmt.Sprintf( `{"device_id":%q,"execution_id":%q,"claim_generation":%d,"command_sha256":%q,"card_signature":%q,"detail_signature":%q,"observed_title":%q,"selected_sku":%q,"quantity":2,"unit_price_cents":2150,"total_price_cents":4300,"evidence_asset_id":%q,"evidence_sha256":%q}`, deviceTestDeviceID, started.Execution.ID, started.Task.ClaimGeneration, command.CommandSHA256, command.Candidate.CardSignature, strings.Repeat("f", 64), command.Candidate.Title, command.OriginalSKU, dryRunEvidenceID, dryRunEvidenceSHA, ) badSKUDryRun := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-dry-runs/" + command.ID + "/ready", contentType: "application/json", body: strings.NewReader(strings.Replace( readyDryRunPayload, fmt.Sprintf(`"selected_sku":%q`, command.OriginalSKU), `"selected_sku":"wrong-sku"`, 1, )), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-dry-run-ready-bad-sku", }) requireDeviceStatus(t, badSKUDryRun, http.StatusConflict) badTotalDryRun := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-dry-runs/" + command.ID + "/ready", contentType: "application/json", body: strings.NewReader(strings.Replace( readyDryRunPayload, `"total_price_cents":4300`, `"total_price_cents":4301`, 1, )), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-dry-run-ready-bad-total", }) requireDeviceStatus(t, badTotalDryRun, http.StatusConflict) readyDryRun := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-dry-runs/" + command.ID + "/ready", contentType: "application/json", body: strings.NewReader(readyDryRunPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-dry-run-ready", }) requireDeviceStatus(t, readyDryRun, http.StatusOK) if !strings.Contains(readyDryRun.Body.String(), `"status":"READY"`) { t.Fatalf("dry-run ready = %s", readyDryRun.Body.String()) } var readyDryRunResponse struct { DryRun struct { ID string `json:"id"` } `json:"dry_run"` } decodeResponse(t, readyDryRun, &readyDryRunResponse) readyDryRunReplay := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-dry-runs/" + command.ID + "/ready", contentType: "application/json", body: strings.NewReader(readyDryRunPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-dry-run-ready", }) requireDeviceStatus(t, readyDryRunReplay, http.StatusOK) if !strings.Contains(readyDryRunReplay.Body.String(), `"replayed":true`) { t.Fatalf("dry-run ready replay = %s", readyDryRunReplay.Body.String()) } startSubmissionPayload := fmt.Sprintf( `{"device_id":%q,"execution_id":%q,"claim_generation":%d,"command_id":%q,"command_sha256":%q,"dry_run_id":%q,"dry_run_evidence_sha256":%q,"observed_title":%q,"selected_sku":%q,"quantity":2,"unit_price_cents":2150,"total_price_cents":4300}`, deviceTestDeviceID, started.Execution.ID, started.Task.ClaimGeneration, command.ID, command.CommandSHA256, readyDryRunResponse.DryRun.ID, dryRunEvidenceSHA, command.Candidate.Title, command.OriginalSKU, ) startSubmission := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-submissions/start", contentType: "application/json", body: strings.NewReader(startSubmissionPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-submission-start", }) requireDeviceStatus(t, startSubmission, http.StatusOK) var submissionResponse struct { Submission struct { ID string `json:"id"` } `json:"submission"` } decodeResponse(t, startSubmission, &submissionResponse) if submissionResponse.Submission.ID == "" || !strings.Contains(startSubmission.Body.String(), `"status":"FENCED"`) { t.Fatalf("submission start = %s", startSubmission.Body.String()) } startSubmissionReplay := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-submissions/start", contentType: "application/json", body: strings.NewReader(startSubmissionPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-submission-start", }) requireDeviceStatus(t, startSubmissionReplay, http.StatusOK) if !strings.Contains(startSubmissionReplay.Body.String(), `"replayed":true`) { t.Fatalf("submission start replay = %s", startSubmissionReplay.Body.String()) } secondSubmission := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-submissions/start", contentType: "application/json", body: strings.NewReader(startSubmissionPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-submission-start-second", }) requireDeviceStatus(t, secondSubmission, http.StatusConflict) manualReviewPayload := fmt.Sprintf( `{"device_id":%q,"execution_id":%q,"claim_generation":%d,"command_id":%q,"command_sha256":%q,"reason_code":"ORDER_PAGE_UNKNOWN"}`, deviceTestDeviceID, started.Execution.ID, started.Task.ClaimGeneration, command.ID, command.CommandSHA256, ) manualReview := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-submissions/" + submissionResponse.Submission.ID + "/manual-review", contentType: "application/json", body: strings.NewReader(manualReviewPayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-submission-manual", }) requireDeviceStatus(t, manualReview, http.StatusOK) if !strings.Contains(manualReview.Body.String(), `"status":"MANUAL_REVIEW"`) { t.Fatalf("submission manual review = %s", manualReview.Body.String()) } const reconciliationEvidenceID = "00000000-0000-4000-8000-000000000078" reconciliationEvidenceSHA := strings.Repeat("d", 64) if _, err := fixture.db.Exec( `INSERT INTO execution_evidence_assets ( id, task_id, execution_id, media_type, size_bytes, sha256, storage_key, created_at, received_after_execution_expiry ) VALUES (?, ?, ?, 'image/jpeg', 10, ?, ?, ?, 0)`, reconciliationEvidenceID, taskID, started.Execution.ID, reconciliationEvidenceSHA, "orders/reconciliation.jpg", time.Now().UTC().Format(time.RFC3339Nano), ); err != nil { t.Fatalf("seed reconciliation evidence: %v", err) } reconcilePayload := fmt.Sprintf( `{"device_id":%q,"execution_id":%q,"claim_generation":%d,"command_id":%q,"command_sha256":%q,"platform_order_no":"12345678901234567890","platform_ordered_at":%q,"platform_order_status":"PENDING_PAYMENT","observed_title":%q,"selected_sku":%q,"quantity":2,"total_price_cents":4300,"evidence_asset_id":%q,"evidence_sha256":%q}`, deviceTestDeviceID, started.Execution.ID, started.Task.ClaimGeneration, command.ID, command.CommandSHA256, time.Now().UTC().Format(time.RFC3339), command.Candidate.Title, command.OriginalSKU, reconciliationEvidenceID, reconciliationEvidenceSHA, ) reconciled := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-submissions/" + submissionResponse.Submission.ID + "/reconcile", contentType: "application/json", body: strings.NewReader(reconcilePayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-submission-reconcile", }) requireDeviceStatus(t, reconciled, http.StatusOK) if !strings.Contains(reconciled.Body.String(), `"status":"RECONCILED"`) || !strings.Contains(reconciled.Body.String(), `"platform_order_status":"PENDING_PAYMENT"`) { t.Fatalf("submission reconciliation = %s", reconciled.Body.String()) } reconciledReplay := performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/order-submissions/" + submissionResponse.Submission.ID + "/reconcile", contentType: "application/json", body: strings.NewReader(reconcilePayload), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "order-submission-reconcile", }) requireDeviceStatus(t, reconciledReplay, http.StatusOK) if !strings.Contains(reconciledReplay.Body.String(), `"replayed":true`) { t.Fatalf("submission reconcile replay = %s", reconciledReplay.Body.String()) } var outcomeSubmitted bool if err := fixture.db.QueryRow( `SELECT order_submitted FROM execution_outcomes WHERE execution_id = ?`, started.Execution.ID, ).Scan(&outcomeSubmitted); err != nil { t.Fatalf("query reconciled outcome: %v", err) } if !outcomeSubmitted { t.Fatal("reconciled execution outcome did not record order_submitted") } var taskStatus, authorizationStatus string if err := fixture.db.QueryRow( `SELECT status FROM purchase_tasks WHERE id = ?`, taskID, ).Scan(&taskStatus); err != nil { t.Fatalf("query reconciled task: %v", err) } if err := fixture.db.QueryRow( `SELECT status FROM order_authorizations WHERE id = ?`, command.ID, ).Scan(&authorizationStatus); err != nil { t.Fatalf("query consumed authorization: %v", err) } if taskStatus != "SUCCEEDED" || authorizationStatus != "CONSUMED" { t.Fatalf( "reconciled task/authorization = %s/%s", taskStatus, authorizationStatus, ) } adminDetail := performAdminRequest( t, fixture.adminRouter, http.MethodGet, "/api/v1/tasks/"+taskID, "", nil, "", ) requireAdminStatus(t, adminDetail, http.StatusOK) if adminDetail.Header().Get("Cache-Control") != "no-store" { t.Fatalf( "admin detail cache control = %q", adminDetail.Header().Get("Cache-Control"), ) } var adminDetailBody struct { OrderSubmissions []struct { ID string `json:"id"` AuthorizationID string `json:"authorization_id"` Status string `json:"status"` ExpectedSKU string `json:"expected_sku"` ExpectedQuantity int `json:"expected_quantity"` ExpectedTotalCents int64 `json:"expected_total_price_cents"` PlatformOrderNo string `json:"platform_order_no"` PlatformOrderedAt string `json:"platform_ordered_at"` PlatformOrderStatus string `json:"platform_order_status"` EvidenceAssetID string `json:"reconciliation_evidence_asset_id"` } `json:"order_submissions"` } decodeResponse(t, adminDetail, &adminDetailBody) if len(adminDetailBody.OrderSubmissions) != 1 { t.Fatalf( "admin order submissions = %+v", adminDetailBody.OrderSubmissions, ) } adminSubmission := adminDetailBody.OrderSubmissions[0] if adminSubmission.ID != submissionResponse.Submission.ID || adminSubmission.AuthorizationID != command.ID || adminSubmission.Status != "RECONCILED" || adminSubmission.ExpectedSKU != command.OriginalSKU || adminSubmission.ExpectedQuantity != 2 || adminSubmission.ExpectedTotalCents != 4300 || adminSubmission.PlatformOrderNo != "12345678901234567890" || adminSubmission.PlatformOrderedAt == "" || adminSubmission.PlatformOrderStatus != "PENDING_PAYMENT" || adminSubmission.EvidenceAssetID != reconciliationEvidenceID { t.Fatalf("admin submission = %+v", adminSubmission) } detailAfterReconcile, err := fixture.tasks.Get( context.Background(), localAdminSubject, taskID, ) if err != nil || len(detailAfterReconcile.OrderSubmissions) != 1 || detailAfterReconcile.OrderSubmissions[0].ID != submissionResponse.Submission.ID { t.Fatalf( "task detail submissions = %+v, error = %v", detailAfterReconcile.OrderSubmissions, err, ) } var deliveredEvents, acknowledgedEvents, dryRunStartedEvents, dryRunReadyEvents, fencedEvents, manualEvents, reconciledEvents int for eventType, target := range map[string]*int{ "ORDER_AUTHORIZATION_DELIVERED": &deliveredEvents, "ORDER_AUTHORIZATION_ACKNOWLEDGED": &acknowledgedEvents, "ORDER_DRY_RUN_STARTED": &dryRunStartedEvents, "ORDER_DRY_RUN_READY": &dryRunReadyEvents, "ORDER_SUBMISSION_FENCED": &fencedEvents, "ORDER_SUBMISSION_MANUAL_REVIEW": &manualEvents, "ORDER_SUBMISSION_RECONCILED": &reconciledEvents, } { if err := fixture.db.QueryRow( `SELECT COUNT(*) FROM task_events WHERE task_id = ? AND event_type = ?`, taskID, eventType, ).Scan(target); err != nil { t.Fatalf("count %s events: %v", eventType, err) } } if deliveredEvents != 1 || acknowledgedEvents != 1 || dryRunStartedEvents != 1 || dryRunReadyEvents != 1 || fencedEvents != 1 || manualEvents != 1 || reconciledEvents != 1 { t.Fatalf( "order workflow events = %d/%d/%d/%d/%d/%d/%d", deliveredEvents, acknowledgedEvents, dryRunStartedEvents, dryRunReadyEvents, fencedEvents, manualEvents, reconciledEvents, ) } runner, err := migration.New(fixture.db) if err != nil { t.Fatalf("migration.New() error = %v", err) } if err := runner.Down(context.Background()); err != nil { t.Fatalf("date sync migration down: %v", err) } if err := runner.Down(context.Background()); err != nil { t.Fatalf("procurement migration down: %v", err) } if err := runner.Down(context.Background()); err != nil { t.Fatalf("freight migration down: %v", err) } if err := runner.Down(context.Background()); err == nil { t.Fatal("order submission migration down succeeded with retained data") } } func seedDeviceOrderCommandCandidate( t *testing.T, fixture *deviceHTTPFixture, taskID string, executionID string, ) string { t.Helper() detail, err := fixture.tasks.Get( context.Background(), localAdminSubject, taskID, ) if err != nil { t.Fatalf("get running task detail: %v", err) } taskHash := usecase.TaskContentSHA256(detail.Task) now := time.Now().UTC().Format(time.RFC3339Nano) candidateKey := strings.Repeat("7", 64) if _, err := fixture.db.Exec( `INSERT INTO candidate_search_runs ( execution_id, task_id, task_content_sha256, execution_mode, search_query, started_at, received_at, observation_count, collection_complete, received_after_execution_expiry ) VALUES (?, ?, ?, 'MANUAL_FIRST', 'PDD_IMAGE_SEARCH', ?, ?, 1, 1, 0)`, executionID, taskID, taskHash, now, now, ); err != nil { t.Fatalf("seed order command search run: %v", err) } if _, err := fixture.db.Exec( `INSERT INTO candidate_observations ( execution_id, task_id, ordinal, title, sku_text, price_text, product_url, image_url, evidence_asset_ids_json, collection_status, observed_at ) VALUES (?, ?, 1, '设备命令候选', 'TEST-SKU-COMMAND', '21.50', '', '', '[]', 'COMPLETE', ?)`, executionID, taskID, now, ); err != nil { t.Fatalf("seed order command observation: %v", err) } if _, err := fixture.db.Exec( `INSERT INTO candidate_observation_identities ( candidate_key, execution_id, candidate_ordinal, card_signature, detail_signature, detail_evidence_sha256, specification_evidence_sha256, identity_version, created_at ) VALUES (?, ?, 1, ?, ?, ?, ?, 1, ?)`, candidateKey, executionID, strings.Repeat("8", 64), strings.Repeat("9", 64), strings.Repeat("a", 64), strings.Repeat("b", 64), now, ); err != nil { t.Fatalf("seed order command identity: %v", err) } if _, err := fixture.db.Exec( `UPDATE purchase_tasks SET status = 'WAITING_CONFIRMATION', version = version + 1, updated_at = ? WHERE id = ? AND status = 'RUNNING'`, now, taskID, ); err != nil { t.Fatalf("move order command task to waiting: %v", err) } if _, err := fixture.db.Exec( `UPDATE task_executions SET current_step = 'WAITING_ADMIN_CONFIRMATION', last_heartbeat_at = ? WHERE id = ?`, now, executionID, ); err != nil { t.Fatalf("move order command execution to waiting: %v", err) } return candidateKey } func TestDeviceReleaseReturnsClaimedTaskToPending(t *testing.T) { fixture := newDeviceHTTPFixture(t) requireDeviceStatus(t, fixture.readyHeartbeat(t), http.StatusOK) taskID := fixture.createPendingTask(t) claim := fixture.claimNext(t, "claim-for-release", testOpaqueToken) requireDeviceStatus(t, claim, http.StatusOK) var claimed deviceLifecycleResponse decodeResponse(t, claim, &claimed) releaseBody := fmt.Sprintf( `{"claim_generation":%d,"expected_version":%d}`, claimed.Task.ClaimGeneration, claimed.Task.Version, ) releasedResponse := performDeviceRequest( t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/release", contentType: "application/json", body: strings.NewReader(releaseBody), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "release-claim", }, ) requireDeviceStatus(t, releasedResponse, http.StatusOK) var released deviceLifecycleResponse decodeResponse(t, releasedResponse, &released) if released.Task.ID != taskID || released.Task.Status != string(domain.TaskStatusPending) || released.Task.Version != claimed.Task.Version+1 || released.Task.ClaimGeneration != claimed.Task.ClaimGeneration || released.Replayed { t.Fatalf("release response = %+v", released) } assertNoClaimSecret(t, releasedResponse, testOpaqueToken) } func TestDeviceCancelAcknowledgementFollowsAdminStopRequest( t *testing.T, ) { fixture := newDeviceHTTPFixture(t) requireDeviceStatus(t, fixture.readyHeartbeat(t), http.StatusOK) taskID := fixture.createPendingTask(t) claim := fixture.claimNext(t, "claim-for-cancel", testOpaqueToken) requireDeviceStatus(t, claim, http.StatusOK) var claimed deviceLifecycleResponse decodeResponse(t, claim, &claimed) startBody := fmt.Sprintf( `{"claim_generation":%d,"expected_version":%d}`, claimed.Task.ClaimGeneration, claimed.Task.Version, ) startResponse := performDeviceRequest( t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/start", contentType: "application/json", body: strings.NewReader(startBody), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "start-for-cancel", }, ) requireDeviceStatus(t, startResponse, http.StatusOK) var started deviceLifecycleResponse decodeResponse(t, startResponse, &started) cancelRequested, err := fixture.tasks.Cancel( context.Background(), usecase.CancelTaskCommand{ CreatorSubject: "local-admin", ActorUserID: deviceTestAdminID, TaskID: taskID, Reason: "HTTP cancel acknowledgement test", }, ) if err != nil || cancelRequested.Status != domain.TaskStatusRunning || cancelRequested.CancelRequestedAt == nil { t.Fatalf( "admin cancel request = %+v, error = %v", cancelRequested, err, ) } heartbeatBody := fmt.Sprintf( `{ "execution_id":%q, "claim_generation":%d, "step":"STOPPING" }`, started.Execution.ID, started.Task.ClaimGeneration, ) heartbeatResponse := performDeviceRequest( t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/heartbeat", contentType: "application/json", body: strings.NewReader(heartbeatBody), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, }, ) requireDeviceStatus(t, heartbeatResponse, http.StatusOK) var heartbeat deviceLifecycleResponse decodeResponse(t, heartbeatResponse, &heartbeat) if !heartbeat.CancelRequested { t.Fatalf("cancel heartbeat response = %+v", heartbeat) } ackBody := fmt.Sprintf( `{ "execution_id":%q, "claim_generation":%d, "expected_version":%d }`, started.Execution.ID, heartbeat.Task.ClaimGeneration, heartbeat.Task.Version, ) acknowledged := performDeviceRequest( t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/cancel-ack", contentType: "application/json", body: strings.NewReader(ackBody), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "cancel-ack-http", }, ) requireDeviceStatus(t, acknowledged, http.StatusOK) var canceled deviceLifecycleResponse decodeResponse(t, acknowledged, &canceled) if canceled.Task.Status != string(domain.TaskStatusCanceled) || canceled.Replayed { t.Fatalf("cancel acknowledgement = %+v", canceled) } assertNoClaimSecret(t, acknowledged, testOpaqueToken) replayed := performDeviceRequest( t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/" + taskID + "/cancel-ack", contentType: "application/json", body: strings.NewReader(ackBody), bearerToken: testOpaqueToken, claimToken: testOpaqueToken, idempotencyKey: "cancel-ack-http", }, ) requireDeviceStatus(t, replayed, http.StatusOK) var replayedBody deviceLifecycleResponse decodeResponse(t, replayed, &replayedBody) if !replayedBody.Replayed || replayedBody.Task.Status != string(domain.TaskStatusCanceled) { t.Fatalf("cancel acknowledgement replay = %+v", replayedBody) } } type deviceHTTPFixture struct { db *sql.DB store *repository.Store assets *usecase.AssetService tasks *usecase.TaskService router http.Handler adminRouter http.Handler taskSequence int } func newDeviceHTTPFixture(t *testing.T) *deviceHTTPFixture { t.Helper() ctx := context.Background() db, err := database.Open( ctx, filepath.Join(t.TempDir(), "device-handlers.db"), ) if err != nil { t.Fatalf("database.Open() error = %v", err) } t.Cleanup(func() { _ = db.Close() }) runner, err := migration.New(db) if err != nil { t.Fatalf("migration.New() error = %v", err) } if _, err := runner.Up(ctx); err != nil { t.Fatalf("migration.Up() error = %v", err) } seedDeviceHTTPIdentity(t, db) store, err := repository.New(db) if err != nil { t.Fatalf("repository.New() error = %v", err) } clock := usecase.SystemClock{} ids := usecase.UUIDGenerator{} tasks, err := usecase.NewTaskService(store, clock, ids) if err != nil { t.Fatalf("usecase.NewTaskService() error = %v", err) } files, err := assetstore.New(filepath.Join(t.TempDir(), "assets")) if err != nil { t.Fatalf("assetstore.New() error = %v", err) } assets, err := usecase.NewAssetService(store, files, clock, ids) if err != nil { t.Fatalf("usecase.NewAssetService() error = %v", err) } lifecycle, err := usecase.NewLifecycleService( store, clock, ids, 10*time.Minute, 10*time.Minute, 2*time.Minute, ) if err != nil { t.Fatalf("usecase.NewLifecycleService() error = %v", err) } results, err := usecase.NewExecutionResultService(store, files, clock, ids) if err != nil { t.Fatalf("usecase.NewExecutionResultService() error = %v", err) } authorizations, err := usecase.NewOrderAuthorizationService( store, clock, ids, ) if err != nil { t.Fatalf("usecase.NewOrderAuthorizationService() error = %v", err) } commands, err := usecase.NewDeviceOrderCommandService(store, clock, ids) if err != nil { t.Fatalf("usecase.NewDeviceOrderCommandService() error = %v", err) } dryRuns, err := usecase.NewOrderDryRunService(store, clock, ids) if err != nil { t.Fatalf("usecase.NewOrderDryRunService() error = %v", err) } submissions, err := usecase.NewOrderSubmissionService(store, clock, ids) if err != nil { t.Fatalf("usecase.NewOrderSubmissionService() error = %v", err) } deviceRoutes, err := NewDeviceRouteRegistrar( DeviceServices{ Lifecycle: lifecycle, Assets: assets, Results: results, Commands: commands, DryRuns: dryRuns, Submissions: submissions, }, ) if err != nil { t.Fatalf("NewDeviceRouteRegistrar() error = %v", err) } authenticator := deviceHTTPAuthenticator{} router, err := NewRouter(RouterDependencies{ Database: db, RegisterPublicRoutes: discardRoutes, RegisterAdminRoutes: registerDeviceTestAdminProbe, RegisterDeviceRoutes: deviceRoutes, AdminSessions: authenticator, DeviceAccess: authenticator, LogEvent: discardEvent, }) if err != nil { t.Fatalf("NewRouter() error = %v", err) } adminRoutes, err := NewAdminRouteRegistrar( AdminServices{ Assets: assets, Tasks: tasks, Results: results, Authorizations: authorizations, }, emptyAdminWeb{}, ) if err != nil { t.Fatalf("NewAdminRouteRegistrar() error = %v", err) } adminRouter, err := NewRouter(RouterDependencies{ Database: db, RegisterPublicRoutes: discardRoutes, RegisterAdminRoutes: adminRoutes, RegisterDeviceRoutes: discardRoutes, AdminSessions: authenticator, DeviceAccess: authenticator, LogEvent: discardEvent, }) if err != nil { t.Fatalf("NewRouter(admin) error = %v", err) } return &deviceHTTPFixture{ db: db, store: store, assets: assets, tasks: tasks, router: router, adminRouter: adminRouter, } } func seedDeviceHTTPIdentity(t *testing.T, db *sql.DB) { t.Helper() now := time.Now().UTC().Add(-time.Minute).Format(time.RFC3339Nano) for _, user := range []struct { id string username string role domain.UserRole }{ {deviceTestAdminID, "device-http-admin", domain.UserRoleAdmin}, {deviceTestBuyerID, "device-http-buyer", domain.UserRoleBuyer}, } { if _, err := db.ExecContext( context.Background(), `INSERT INTO users ( id, username, password_hash, role, is_active, created_at, updated_at ) VALUES (?, ?, 'test-only-password-hash', ?, 1, ?, ?)`, user.id, user.username, user.role, now, now, ); err != nil { t.Fatalf("seed user %s: %v", user.username, err) } } if _, err := db.ExecContext( context.Background(), `INSERT INTO devices ( id, name, token_hash, bound_user_id, is_enabled, created_at, updated_at ) VALUES (?, 'device-http-test', ?, ?, 1, ?, ?)`, deviceTestDeviceID, deviceTestHash("device-registration-token"), deviceTestBuyerID, now, now, ); err != nil { t.Fatalf("seed device: %v", err) } } func (fixture *deviceHTTPFixture) readyHeartbeat( t *testing.T, ) *httptest.ResponseRecorder { t.Helper() return performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/devices/heartbeat", contentType: "application/json", body: strings.NewReader(`{ "app_version":"0.1.0", "android_version":"16", "pdd_version":"8.17.0", "readiness":{ "accessibility_enabled":true, "pdd_installed":true, "active_task_id":null } }`), bearerToken: testOpaqueToken, }) } func (fixture *deviceHTTPFixture) claimNext( t *testing.T, idempotencyKey string, claimToken string, ) *httptest.ResponseRecorder { t.Helper() return performDeviceRequest(t, fixture.router, deviceRequest{ method: http.MethodPost, target: "/api/v1/tasks/claim-next", contentType: "application/json", body: strings.NewReader(`{}`), bearerToken: testOpaqueToken, claimToken: claimToken, idempotencyKey: idempotencyKey, }) } func (fixture *deviceHTTPFixture) createPendingTask(t *testing.T) string { t.Helper() fixture.taskSequence++ index := fixture.taskSequence uploaded, err := fixture.assets.UploadTaskReference( context.Background(), usecase.UploadTaskReferenceCommand{ CreatorSubject: "local-admin", IdempotencyKey: fmt.Sprintf("device-http-asset-%d", index), DeclaredMediaType: "image/jpeg", Content: deviceReferenceImage(t, index), }, ) if err != nil || uploaded.Asset.ID == "" { t.Fatalf("upload asset = %+v, error = %v", uploaded, err) } sourceRef := fmt.Sprintf("device-http-source-%d", index) result, err := fixture.tasks.Create( context.Background(), usecase.CreateTaskCommand{ CreatorSubject: "local-admin", ActorUserID: deviceTestAdminID, IdempotencyKey: fmt.Sprintf("device-http-task-%d", index), SourceRef: &sourceRef, Title: "测试采购商品", Description: "仅用于 HTTP 合同测试", SKU: fmt.Sprintf("TEST-SKU-%d", index), ImageAssetID: uploaded.Asset.ID, Quantity: 2, }, ) if err != nil { t.Fatalf("create task: %v", err) } return result.Task.ID } func deviceReferenceImage(t *testing.T, index int) io.Reader { t.Helper() var encoded bytes.Buffer source := image.NewRGBA(image.Rect(0, 0, 4, 4)) for y := 0; y < 4; y++ { for x := 0; x < 4; x++ { source.Set(x, y, color.RGBA{ R: uint8(40 + index), G: uint8(80 + x), B: uint8(120 + y), A: 255, }) } } if err := jpeg.Encode(&encoded, source, nil); err != nil { t.Fatalf("encode device reference image: %v", err) } return bytes.NewReader(encoded.Bytes()) } type deviceRequest struct { method string target string contentType string body io.Reader bearerToken string claimToken string idempotencyKey string executionID string claimGeneration int64 } func performDeviceRequest( t *testing.T, router http.Handler, spec deviceRequest, ) *httptest.ResponseRecorder { t.Helper() request := httptest.NewRequest(spec.method, spec.target, spec.body) if spec.contentType != "" { request.Header.Set("Content-Type", spec.contentType) } if spec.bearerToken != "" { request.Header.Set( "Authorization", "Bearer "+spec.bearerToken, ) } if spec.claimToken != "" { request.Header.Set(claimTokenHeader, spec.claimToken) } if spec.idempotencyKey != "" { request.Header.Set("Idempotency-Key", spec.idempotencyKey) } if spec.executionID != "" { request.Header.Set("X-Execution-ID", spec.executionID) request.Header.Set( "X-Claim-Generation", strconv.FormatInt(spec.claimGeneration, 10), ) } response := httptest.NewRecorder() router.ServeHTTP(response, request) return response } type deviceLifecycleResponse struct { Task struct { ID string `json:"id"` Status string `json:"status"` Version int64 `json:"version"` ClaimGeneration int64 `json:"claim_generation"` ReferenceImageURL string `json:"reference_image_url"` } `json:"task"` Execution struct { ID string `json:"id"` CurrentStep string `json:"current_step"` OrderSubmitted bool `json:"order_submitted"` ExpiresAt time.Time `json:"execution_expires_at"` } `json:"execution"` Replayed bool `json:"replayed"` CancelRequested bool `json:"cancel_requested"` } func requireDeviceStatus( t *testing.T, response *httptest.ResponseRecorder, want int, ) { t.Helper() if response.Code != want { t.Fatalf( "status/body = %d / %s, want %d", response.Code, response.Body, want, ) } } func assertNoClaimSecret( t *testing.T, response *httptest.ResponseRecorder, claimToken string, ) { t.Helper() responseText := strings.ToLower(response.Body.String()) claimHash := deviceTestHash(claimToken) for _, forbidden := range []string{ strings.ToLower(claimToken), claimHash, "x-claim-token", "claim_token", "claim_token_hash", } { if strings.Contains(responseText, forbidden) { t.Fatalf( "response exposes claim secret %q: %s", forbidden, response.Body, ) } } if response.Header().Get(claimTokenHeader) != "" { t.Fatalf( "%s response header must be empty", claimTokenHeader, ) } } func deviceTestHash(value string) string { sum := sha256.Sum256([]byte(value)) return hex.EncodeToString(sum[:]) } type deviceHTTPAuthenticator struct{} func (deviceHTTPAuthenticator) AuthenticateAdmin( context.Context, string, ) (domain.AuthPrincipal, error) { return domain.AuthPrincipal{ UserID: deviceTestAdminID, Username: "device-http-admin", Role: domain.UserRoleAdmin, SessionID: "device-http-admin-session", ExpiresAt: time.Now().Add(time.Hour), }, nil } func (deviceHTTPAuthenticator) AuthenticateAccessToken( _ context.Context, token string, ) (domain.AuthPrincipal, error) { switch token { case testOpaqueToken: return domain.AuthPrincipal{ UserID: deviceTestBuyerID, Username: "device-http-buyer", Role: domain.UserRoleBuyer, DeviceID: deviceTestDeviceID, ExpiresAt: time.Now().Add( time.Hour, ), }, nil case testCSRFOpaqueToken: return domain.AuthPrincipal{ UserID: deviceTestAdminID, Username: "device-http-admin", Role: domain.UserRoleAdmin, SessionID: "device-http-admin-session", ExpiresAt: time.Now().Add( time.Hour, ), }, nil default: return domain.AuthPrincipal{}, errors.New( "test access token is invalid", ) } } func registerDeviceTestAdminProbe(routes gin.IRoutes) error { routes.GET("/api/v1/admin-probe", func(ctx *gin.Context) { ctx.Status(http.StatusNoContent) }) return nil } var ( _ AdminAuthenticator = deviceHTTPAuthenticator{} _ DeviceAuthenticator = deviceHTTPAuthenticator{} )