package service import ( "database/sql" "encoding/json" "errors" "fmt" "strings" "cmautobuy/admin/model" "cmautobuy/admin/repository" ) // 提交结果时可能出现的几种业务错误,handler 据此决定 HTTP 状态码。 var ( // ErrTaskNotFound 任务不存在 -> 404 ErrTaskNotFound = errors.New("任务不存在") // ErrNeverClaimed 这台客户端从没领过这个任务 -> 403。 // 注意**只有这一种情况才拒绝**,见 SubmitResult 的说明。 ErrNeverClaimed = errors.New("该任务从未分配给这个客户端") // ErrIdempotencyConflict 同一个键提交了不同内容 -> 409 ErrIdempotencyConflict = repository.ErrIdempotencyConflict ) // ResultRequest 是客户端提交成功结果的请求体。 type ResultRequest struct { TaskVersion int `json:"task_version"` AttemptID string `json:"attempt_id"` ResultType string `json:"result_type"` CompletedAt string `json:"completed_at"` PddData json.RawMessage `json:"pdd_data"` } // FailureRequest 是客户端提交失败/需人工处理的请求体。 type FailureRequest struct { TaskVersion int `json:"task_version"` AttemptID string `json:"attempt_id"` Status string `json:"status"` Error struct { Code string `json:"code"` Message string `json:"message"` Retryable bool `json:"retryable"` Step string `json:"step"` } `json:"error"` Diagnostics json.RawMessage `json:"diagnostics"` ReportedAt string `json:"reported_at"` } // SubmitResult 接收客户端提交的成功结果,返回要回给客户端的 JSON。 // // # 最容易写错的一条:无条件接受 // // 下面每一条都不允许违反(docs/admin/04-client-api.md §4.1): // // - **不得**因为任务已取消而拒绝; // - **不得**因为任务已重派给别的客户端而拒绝; // - **必须**能接受同一任务来自多个客户端的多份结果; // - 只有**从未领过**这个任务的客户端才返回 403。 // // 原因:客户端中途不查任务状态(这是有意的设计),所以它**必然**会提交 // 一些"Admin 这边已经不要了"的结果。而它可能真的已经在拼多多下过单了, // 这些数据必须能交上来留痕,否则就成了一笔谁都不知道的订单。 // // accepted: true 的意思是"**我收到并存下了**",不代表这个任务还算数。 // 任务算不算数由人工审核决定。 func SubmitResult(db *sql.DB, taskID, clientID, idemKey string, rawBody []byte) (string, error) { var req ResultRequest if err := json.Unmarshal(rawBody, &req); err != nil { return "", fmt.Errorf("请求体不是合法 JSON: %w", err) } return submitInTx(db, taskID, clientID, idemKey, rawBody, func(tx *sql.Tx, taskType model.TaskType, goodsID string) (map[string]any, error) { pddData := string(req.PddData) if strings.TrimSpace(pddData) == "" { pddData = "{}" } if err := repository.MarkTaskSucceeded(tx, taskID, pddData); err != nil { return nil, err } // 采集任务的结果还要落到商品级,供后续规格匹配使用 if taskType == model.TaskCollect { if err := repository.SetCollectResult(tx, goodsID, pddData); err != nil { return nil, err } } return map[string]any{ "accepted": true, "result_id": newID(), "accepted_at": model.NowISO(), }, nil }) } // SubmitFailure 接收客户端提交的失败/需人工处理结果。 // // §4.1 的无条件接受规则**同样适用于本函数**。 func SubmitFailure(db *sql.DB, taskID, clientID, idemKey string, rawBody []byte) (string, error) { var req FailureRequest if err := json.Unmarshal(rawBody, &req); err != nil { return "", fmt.Errorf("请求体不是合法 JSON: %w", err) } newStatus, ok := mapFailureStatus(req.Status) if !ok { return "", fmt.Errorf( "status 只能是 retry_wait / manual_review / failed / cancelled,收到 %q", req.Status) } return submitInTx(db, taskID, clientID, idemKey, rawBody, func(tx *sql.Tx, taskType model.TaskType, goodsID string) (map[string]any, error) { if err := repository.MarkTaskFailure( tx, taskID, newStatus, req.Error.Code, req.Error.Message); err != nil { return nil, err } // 采集失败要让操作员在蝦皮数据页看得见原因 if taskType == model.TaskCollect { msg := req.Error.Message if req.Error.Code != "" { msg = req.Error.Code + ": " + msg } if err := repository.SetCollectFailed(tx, goodsID, msg); err != nil { return nil, err } } return map[string]any{ "accepted": true, "task_status": string(newStatus), "accepted_at": model.NowISO(), }, nil }) } // mapFailureStatus 把客户端报告的状态映射成 Admin 侧状态。 // 映射表见 docs/admin/04-client-api.md §5。 func mapFailureStatus(reported string) (model.TaskStatus, bool) { switch reported { case "retry_wait": // 放回待领取,等客户端下次再来领 return model.TaskAssigned, true case "manual_review": return model.TaskManualReview, true case "failed": return model.TaskFailed, true case "cancelled": return model.TaskCancelled, true default: return "", false } } // submitInTx 把两个提交接口共同的骨架抽出来: // 幂等检查 → 权限判断 → 业务写入 → 记幂等 → 刷新客户端活动时间, // 全部在**一个事务**里完成。 // // 业务写入部分由 apply 提供,它拿到的 tx 和外层是同一个。 func submitInTx( db *sql.DB, taskID, clientID, idemKey string, rawBody []byte, apply func(tx *sql.Tx, taskType model.TaskType, goodsID string) (map[string]any, error), ) (string, error) { hash := repository.HashRequest(rawBody) tx, err := db.Begin() if err != nil { return "", fmt.Errorf("开始事务失败: %w", err) } defer tx.Rollback() // 已提交的事务再 Rollback 是空操作,安全 // 1. 处理过就原样返回上次的响应,绝不重复落库 if body, done, err := repository.LookupIdempotent(tx, idemKey, hash); err != nil { return "", err } else if done { return body, nil } // 2. 任务得存在 exists, taskType, goodsID, err := repository.TaskExists(tx, taskID) if err != nil { return "", err } if !exists { return "", ErrTaskNotFound } // 3. 权限:**只有从没领过的才拒绝**。 // 任务已取消、已重派给别人,都照样接受。 claimed, err := repository.HasEverClaimed(tx, taskID, clientID) if err != nil { return "", err } if !claimed { return "", ErrNeverClaimed } // 4. 业务写入 resp, err := apply(tx, taskType, goodsID) if err != nil { return "", err } respJSON, err := json.Marshal(resp) if err != nil { return "", fmt.Errorf("序列化响应失败: %w", err) } // 5. 记下幂等键。和业务写入在同一个事务里—— // 分开写的话,业务写成功但幂等没记上,客户端重试会被重复处理。 if err := repository.SaveIdempotent(tx, idemKey, hash, string(respJSON)); err != nil { return "", err } // 6. 刷新客户端活动时间。三个接口都要做, // 只在 claim 里做的话,长任务期间会被误判成离线。 if err := repository.TouchClient(tx, clientID); err != nil { return "", err } if err := tx.Commit(); err != nil { return "", fmt.Errorf("提交事务失败: %w", err) } return string(respJSON), nil }