feat(agent): implement #91 controlled recollection

2026-08-26 10:17:45 +08:00
parent 77fdbebc2d
commit 7b6cc00503
11 changed files with 349 additions and 8 deletions
@@ -0,0 +1,124 @@
+package task
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"go-admin/app/goauto/device"
+	"go-admin/app/goauto/models"
+
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+)
+
+func TestAgentResetIsDeviceScopedIdleAndIdempotent(t *testing.T) {
+	db := openTaskDatabase(t)
+	deviceA, tokenA := registerTaskDevice(t, db, "device-a")
+	_, tokenB := registerTaskDevice(t, db, "device-b")
+	target := createTask(t, db, &deviceA.ID)
+	oldTitle := "旧结果"
+	if err := db.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}).Where("id = ?", target.ID).Updates(map[string]any{
+		"status": models.TaskStatusFailed, "active_slot": nil, "title": oldTitle,
+		"error_code": "RULE_NOT_MATCHED", "error_message": "没有找到规格入口",
+	}).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(&models.CollectionColorPrice{TaskID: target.ID, Color: "黑色", PriceCent: 1000}).Error; err != nil {
+		t.Fatal(err)
+	}
+	service := newTaskService(db)
+	request := ActionRequest{RequestID: uuid.NewString()}
+	_, err := service.ResetForDevice(context.Background(), target.ID, request, "")
+	var deviceError *device.ServiceError
+	if !errors.As(err, &deviceError) || deviceError.Code != device.CodeTokenInvalid {
+		t.Fatalf("missing token was accepted: %v", err)
+	}
+	if _, err := service.ResetForDevice(context.Background(), target.ID, request, tokenB); taskErrorCode(t, err) != CodeTaskNotFound {
+		t.Fatalf("other device reset task: %v", err)
+	}
+
+	busy := createTask(t, db, &deviceA.ID)
+	one := uint8(1)
+	if err := db.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}).Where("id = ?", busy.ID).Updates(map[string]any{
+		"status": models.TaskStatusRunning, "active_slot": one, "device_run_slot": one,
+	}).Error; err != nil {
+		t.Fatal(err)
+	}
+	if _, err := service.ResetForDevice(context.Background(), target.ID, request, tokenA); taskErrorCode(t, err) != CodeDeviceBusy {
+		t.Fatalf("busy device reset task: %v", err)
+	}
+	if err := db.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}).Where("id = ?", busy.ID).Updates(map[string]any{
+		"status": models.TaskStatusFailed, "active_slot": nil, "device_run_slot": nil,
+	}).Error; err != nil {
+		t.Fatal(err)
+	}
+	var product models.PDDProduct
+	if err := db.First(&product, target.PDDProductID).Error; err != nil {
+		t.Fatal(err)
+	}
+	purchaseBusy := models.PurchaseTask{
+		PDDProductID: product.ID, DeviceID: &deviceA.ID,
+		ExecutionMode: models.PurchaseExecutionModeRehearsal, Status: models.PurchaseTaskStatusRunning,
+		PDDURLSnapshot: product.URL, PDDGoodsIDSnapshot: product.GoodsID,
+		Quantity: 1, ReferenceUnitPriceCent: 100, MinUnitPriceCent: 20, MaxUnitPriceCent: 150,
+		Currency: "CNY", RuleType: "pddPurchase", RuleSchemaVersion: 1,
+		RequiredCapabilitiesJSON: `[]`, RuleSnapshot: `{}`, CreateRequestID: uuid.NewString(),
+	}
+	if err := db.Create(&purchaseBusy).Error; err != nil {
+		t.Fatal(err)
+	}
+	if _, err := service.ResetForDevice(context.Background(), target.ID, request, tokenA); taskErrorCode(t, err) != CodeDeviceBusy {
+		t.Fatalf("running purchase did not block reset: %v", err)
+	}
+	if err := db.Session(&gorm.Session{SkipHooks: true}).Model(&models.PurchaseTask{}).Where("id = ?", purchaseBusy.ID).Updates(map[string]any{
+		"status": models.PurchaseTaskStatusFailed, "active_slot": nil, "device_run_slot": nil, "account_run_slot": nil,
+	}).Error; err != nil {
+		t.Fatal(err)
+	}
+
+	reset, err := service.ResetForDevice(context.Background(), target.ID, request, tokenA)
+	if err != nil || reset.TaskID != target.ID || reset.Status != models.TaskStatusPending || reset.Replayed {
+		t.Fatalf("reset: %+v %v", reset, err)
+	}
+	var updated models.CollectionTask
+	if err := db.First(&updated, target.ID).Error; err != nil {
+		t.Fatal(err)
+	}
+	if updated.DeviceID == nil || *updated.DeviceID != deviceA.ID || updated.Title != nil || updated.ErrorCode != nil {
+		t.Fatalf("reset did not preserve assignment or clear result: %+v", updated)
+	}
+	var prices int64
+	if err := db.Model(&models.CollectionColorPrice{}).Where("task_id = ?", target.ID).Count(&prices).Error; err != nil || prices != 0 {
+		t.Fatalf("old prices remain: count=%d err=%v", prices, err)
+	}
+	replayed, err := service.ResetForDevice(context.Background(), target.ID, request, tokenA)
+	if err != nil || !replayed.Replayed {
+		t.Fatalf("reset replay: %+v %v", replayed, err)
+	}
+	if _, err := service.ResetForDevice(context.Background(), target.ID, ActionRequest{RequestID: uuid.NewString()}, tokenA); taskErrorCode(t, err) != CodeTaskStateConflict {
+		t.Fatalf("new request reset pending task: %v", err)
+	}
+}
+
+func TestAgentResetRejectsActiveProductTask(t *testing.T) {
+	db := openTaskDatabase(t)
+	deviceRecord, token := registerTaskDevice(t, db, "device-one")
+	target := createTask(t, db, &deviceRecord.ID)
+	if err := db.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}).Where("id = ?", target.ID).Updates(map[string]any{
+		"status": models.TaskStatusCompleted, "active_slot": nil,
+	}).Error; err != nil {
+		t.Fatal(err)
+	}
+	conflict := models.CollectionTask{
+		PDDProductID: target.PDDProductID, RuleID: target.RuleID, DeviceID: &deviceRecord.ID,
+		Status: models.TaskStatusPending, URLSnapshot: target.URLSnapshot,
+		GoodsIDSnapshot: target.GoodsIDSnapshot, RuleSnapshot: target.RuleSnapshot,
+	}
+	if err := db.Create(&conflict).Error; err != nil {
+		t.Fatal(err)
+	}
+	if _, err := newTaskService(db).ResetForDevice(context.Background(), target.ID, ActionRequest{RequestID: uuid.NewString()}, token); taskErrorCode(t, err) != CodeProductTaskActive {
+		t.Fatalf("active product task was ignored: %v", err)
+	}
+}
@@ -80,6 +80,30 @@ func (handler Handler) AgentHistoryDetail(context *gin.Context) {
 	context.JSON(http.StatusOK, gin.H{"data": result})
 }

+func (handler Handler) AgentReset(context *gin.Context) {
+	id, err := taskID(context)
+	if err != nil || id == 0 {
+		writeError(context, serviceError(device.CodeInvalidRequest, "taskId 无效"))
+		return
+	}
+	var request ActionRequest
+	if err := decodeStrict(context, &request); err != nil {
+		writeError(context, serviceError(device.CodeInvalidRequest, "请求 JSON 无效"))
+		return
+	}
+	service, token, ok := handler.service(context)
+	if !ok {
+		return
+	}
+	result, err := service.ResetForDevice(context.Request.Context(), id, request, token)
+	if err != nil {
+		writeError(context, err)
+		return
+	}
+	context.Header("Cache-Control", "no-store")
+	context.JSON(http.StatusOK, gin.H{"data": result})
+}
+
 func (handler Handler) Claim(context *gin.Context) { handler.action(context, (*Service).Claim) }
 func (handler Handler) Start(context *gin.Context) { handler.action(context, (*Service).Start) }

@@ -3,7 +3,9 @@ package task
 import (
 	"context"
 	"errors"
+	"time"

+	"go-admin/app/goauto/device"
 	"go-admin/app/goauto/models"

 	"gorm.io/gorm"
@@ -16,12 +18,44 @@ type DeleteResponse struct {
 	Replayed bool   `json:"replayed,omitempty"`
 }

+type AgentResetResponse struct {
+	TaskID   uint64 `json:"taskId"`
+	Status   string `json:"status"`
+	Replayed bool   `json:"replayed,omitempty"`
+}
+
 func (service *Service) Reset(ctx context.Context, taskID uint64, request ActionRequest) (DetailResponse, error) {
+	return service.reset(ctx, taskID, request, nil)
+}
+
+func (service *Service) ResetForDevice(ctx context.Context, taskID uint64, request ActionRequest, token string) (AgentResetResponse, error) {
+	detail, err := service.reset(ctx, taskID, request, &token)
+	if err != nil {
+		return AgentResetResponse{}, err
+	}
+	return AgentResetResponse{TaskID: detail.Task.ID, Status: detail.Task.Status, Replayed: detail.Replayed}, nil
+}
+
+func (service *Service) reset(ctx context.Context, taskID uint64, request ActionRequest, deviceToken *string) (DetailResponse, error) {
 	if err := validateAction(taskID, request); err != nil {
 		return DetailResponse{}, err
 	}
 	var replayed bool
 	err := service.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var deviceRecord *models.AgentDevice
+		if deviceToken != nil {
+			authenticated, err := device.NewService(tx).Authenticate(ctx, *deviceToken)
+			if err != nil {
+				return err
+			}
+			if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&authenticated, authenticated.ID).Error; err != nil {
+				return internalError(err)
+			}
+			if authenticated.Status != models.DeviceStatusOnline {
+				return serviceError(CodeDeviceOffline, "设备离线，不能重新采集")
+			}
+			deviceRecord = &authenticated
+		}
 		var record models.CollectionTask
 		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&record, taskID).Error; err != nil {
 			if errors.Is(err, gorm.ErrRecordNotFound) {
@@ -29,6 +63,9 @@ func (service *Service) Reset(ctx context.Context, taskID uint64, request Action
 			}
 			return internalError(err)
 		}
+		if deviceRecord != nil && (record.DeviceID == nil || *record.DeviceID != deviceRecord.ID) {
+			return serviceError(CodeTaskNotFound, "采集任务不存在")
+		}
 		if record.ResetRequestID != nil && *record.ResetRequestID == request.RequestID {
 			replayed = true
 			return nil
@@ -47,6 +84,11 @@ func (service *Service) Reset(ctx context.Context, taskID uint64, request Action
 		if active > 0 {
 			return serviceError(CodeProductTaskActive, "该商品已有待执行或执行中的任务")
 		}
+		if deviceRecord != nil {
+			if err := ensureDeviceIdleForReset(tx, deviceRecord.ID, record.ID, service.Now()); err != nil {
+				return err
+			}
+		}
 		if err := deleteResultChildren(tx, record.ID); err != nil {
 			return err
 		}
@@ -71,6 +113,29 @@ func (service *Service) Reset(ctx context.Context, taskID uint64, request Action
 	return detail, err
 }

+func ensureDeviceIdleForReset(tx *gorm.DB, deviceID, taskID uint64, now time.Time) error {
+	var busy int64
+	if err := tx.Model(&models.CollectionTask{}).
+		Where("id <> ? AND device_id = ? AND (status = ? OR (status = ? AND lease_expires_at > ?))", taskID, deviceID, models.TaskStatusRunning, models.TaskStatusPending, now).
+		Count(&busy).Error; err != nil {
+		return internalError(err)
+	}
+	if busy > 0 {
+		return serviceError(CodeDeviceBusy, "设备正在执行其他任务")
+	}
+	if err := tx.Model(&models.PurchaseTask{}).
+		Where("device_id = ? AND (status IN ? OR (status IN ? AND lease_expires_at > ?))", deviceID,
+			[]string{models.PurchaseTaskStatusRunning, models.PurchaseTaskStatusOrderSubmitStarted},
+			[]string{models.PurchaseTaskStatusPending, models.PurchaseTaskStatusSpecProbePending}, now).
+		Count(&busy).Error; err != nil {
+		return internalError(err)
+	}
+	if busy > 0 {
+		return serviceError(CodeDeviceBusy, "设备正在执行其他任务")
+	}
+	return nil
+}
+
 func (service *Service) DeleteFailed(ctx context.Context, taskID uint64, request ActionRequest) (DeleteResponse, error) {
 	if err := validateAction(taskID, request); err != nil {
 		return DeleteResponse{}, err
@@ -19,6 +19,7 @@ func InitRouter(engine *gin.Engine, auth *jwt.GinJWTMiddleware) {
 	agent.GET("/tasks/next", handler.Next)
 	agent.GET("/collection-tasks", handler.AgentHistory)
 	agent.GET("/collection-tasks/:taskId", handler.AgentHistoryDetail)
+	agent.POST("/collection-tasks/:taskId/reset", handler.AgentReset)
 	agent.POST("/tasks/:taskId/claim", handler.Claim)
 	agent.POST("/tasks/:taskId/start", handler.Start)
 	agent.POST("/tasks/:taskId/result", handler.Result)