chengma/cmroubao
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(t216): deliver recoverable device order commands

Browse Source
This commit is contained in:
QiuSW
2026-07-28 13:23:18 +08:00
parent 75fdc63db2
commit 2372ab2280
32 changed files with 2415 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-api/internal/transport/httpapi/admin_handlers.go
+3
View File
@@ -484,6 +484,9 @@ func orderAuthorizationResponse(
"revoked_at": formatOptionalTime(authorization.RevokedAt),
"failure_code": authorization.FailureCode,
"failure_message": authorization.FailureMessage,
"command_sha256": authorization.CommandSHA256,
"delivery_attempt_count": authorization.DeliveryAttemptCount,
"last_delivered_at": formatOptionalTime(authorization.LastDeliveredAt),
}
}
backend-api/internal/transport/httpapi/admin_handlers_test.go
+3
View File
@@ -410,6 +410,9 @@ func TestAdminOrderAuthorizationIsIdempotentAndRevisioned(t *testing.T) {
if err != nil {
t.Fatalf("migration.New() error = %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 authorization migration down succeeded with retained data")
}
backend-api/internal/transport/httpapi/device_handlers.go
+116 -1
View File
@@ -20,10 +20,12 @@ type DeviceServices struct {
Lifecycle *usecase.LifecycleService
Assets *usecase.AssetService
Results *usecase.ExecutionResultService
Commands *usecase.DeviceOrderCommandService
}
func (services DeviceServices) validate() error {
if services.Lifecycle == nil || services.Assets == nil || services.Results == nil {
if services.Lifecycle == nil || services.Assets == nil ||
services.Results == nil || services.Commands == nil {
return errors.New("device services are required")
}
return nil
@@ -73,12 +75,125 @@ func NewDeviceRouteRegistrar(
routes.POST("/api/v1/tasks/:id/evidence", handler.uploadEvidence)
routes.POST("/api/v1/tasks/:id/candidates", handler.storeCandidates)
routes.POST("/api/v1/tasks/:id/human-reviews", handler.storeHumanReview)
routes.POST(
"/api/v1/tasks/:id/commands/next",
handler.pullOrderCommand,
)
routes.POST(
"/api/v1/tasks/:id/commands/:command_id/ack",
handler.acknowledgeOrderCommand,
)
routes.POST("/api/v1/tasks/:id/complete", handler.completeTask)
routes.POST("/api/v1/tasks/:id/fail", handler.failTask)
return nil
}, nil
}
func (handler *deviceHandlers) pullOrderCommand(ctx *gin.Context) {
principal, ok := devicePrincipal(ctx)
if !ok {
return
}
var request struct {
DeviceID string `json:"device_id"`
ExecutionID string `json:"execution_id"`
ClaimGeneration int64 `json:"claim_generation"`
}
if !decodeDeviceJSON(ctx, &request) ||
!deviceIDMatches(ctx, request.DeviceID, principal.DeviceID) {
return
}
command, err := handler.services.Commands.Pull(
ctx.Request.Context(),
usecase.PullDeviceOrderCommand{
UserID: principal.UserID,
DeviceID: principal.DeviceID,
TaskID: ctx.Param("id"),
ExecutionID: request.ExecutionID,
ClaimGeneration: request.ClaimGeneration,
ClaimToken: ctx.GetHeader(claimTokenHeader),
},
)
if err != nil {
writeUsecaseError(ctx, err)
return
}
ctx.Header("Cache-Control", "no-store")
if command == nil {
ctx.Status(http.StatusNoContent)
return
}
ctx.JSON(http.StatusOK, deviceOrderCommandResponse(*command))
}
func (handler *deviceHandlers) acknowledgeOrderCommand(ctx *gin.Context) {
principal, ok := devicePrincipal(ctx)
if !ok {
return
}
var request struct {
DeviceID string `json:"device_id"`
ExecutionID string `json:"execution_id"`
ClaimGeneration int64 `json:"claim_generation"`
CommandSHA256 string `json:"command_sha256"`
}
if !decodeDeviceJSON(ctx, &request) ||
!deviceIDMatches(ctx, request.DeviceID, principal.DeviceID) {
return
}
result, err := handler.services.Commands.Acknowledge(
ctx.Request.Context(),
usecase.AcknowledgeDeviceOrderCommand{
UserID: principal.UserID,
DeviceID: principal.DeviceID,
TaskID: ctx.Param("id"),
ExecutionID: request.ExecutionID,
AuthorizationID: ctx.Param("command_id"),
ClaimGeneration: request.ClaimGeneration,
ClaimToken: ctx.GetHeader(claimTokenHeader),
CommandSHA256: request.CommandSHA256,
IdempotencyKey: ctx.GetHeader("Idempotency-Key"),
},
)
if err != nil {
writeUsecaseError(ctx, err)
return
}
ctx.Header("Cache-Control", "no-store")
ctx.JSON(http.StatusOK, gin.H{
"command_id": result.Authorization.ID,
"status": result.Authorization.Status,
"replayed": result.Replayed,
})
}
func deviceOrderCommandResponse(command domain.DeviceOrderCommand) gin.H {
return gin.H{
"id": command.ID,
"schema_version": command.SchemaVersion,
"type": command.Type,
"authorization_version": command.AuthorizationVersion,
"task_id": command.TaskID,
"execution_id": command.ExecutionID,
"task_content_sha256": command.TaskContentSHA256,
"original_sku": command.OriginalSKU,
"quantity": command.Quantity,
"candidate": gin.H{
"candidate_key": command.CandidateKey,
"observed_ordinal": command.ObservedOrdinal,
"title": command.CandidateTitle,
"sku_text": command.CandidateSKUText,
"price_text": command.CandidatePriceText,
"card_signature": command.CardSignature,
"detail_signature": command.DetailSignature,
"detail_evidence_sha256": command.DetailEvidenceSHA256,
"specification_evidence_sha256": command.SpecificationEvidenceSHA256,
},
"command_sha256": command.CommandSHA256,
"authorization_status": command.AuthorizationStatus,
}
}
func (handler *deviceHandlers) referenceImage(ctx *gin.Context) {
principal, ok := devicePrincipal(ctx)
if !ok {
backend-api/internal/transport/httpapi/device_handlers_test.go
+333
View File
@@ -713,9 +713,17 @@ func TestDeviceExecutionResultsAreIdempotentAndAuditable(t *testing.T) {
if err != nil {
t.Fatalf("migration.New() after review error = %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 != 1 {
t.Fatalf("restored migrations = %d, want 1", 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}`,
@@ -784,6 +792,326 @@ func TestDeviceExecutionResultsAreIdempotentAndAuditable(t *testing.T) {
}
}
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"`
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.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())
}
var deliveredEvents, acknowledgedEvents int
for eventType, target := range map[string]*int{
"ORDER_AUTHORIZATION_DELIVERED": &deliveredEvents,
"ORDER_AUTHORIZATION_ACKNOWLEDGED": &acknowledgedEvents,
} {
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 {
t.Fatalf(
"delivery/ack events = %d/%d",
deliveredEvents,
acknowledgedEvents,
)
}
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.Fatal("device command migration down succeeded with command 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)
@@ -1020,11 +1348,16 @@ func newDeviceHTTPFixture(t *testing.T) *deviceHTTPFixture {
if err != nil {
t.Fatalf("usecase.NewExecutionResultService() error = %v", err)
}
commands, err := usecase.NewDeviceOrderCommandService(store, clock, ids)
if err != nil {
t.Fatalf("usecase.NewDeviceOrderCommandService() error = %v", err)
}
deviceRoutes, err := NewDeviceRouteRegistrar(
DeviceServices{
Lifecycle: lifecycle,
Assets: assets,
Results: results,
Commands: commands,
},
)
if err != nil {