309 lines
11 KiB
Go
309 lines
11 KiB
Go
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
|
|
// ErrCollectMismatch 客户端采回来的商品不是请求的那个 -> 422
|
|
ErrCollectMismatch = errors.New("采集结果与请求的商品不一致")
|
|
)
|
|
|
|
// collectedData 是采集结果里 Admin 关心的那几项。
|
|
//
|
|
// 客户端提交的完整内容会原样存进 tasks.result_data，这里只挑业务要用的解析出来。
|
|
// 用不到的字段不写进结构体，多余的 JSON 字段会被忽略，不影响向前兼容。
|
|
type collectedData struct {
|
|
GoodsID string `json:"goods_id"`
|
|
Title string `json:"title"`
|
|
ShopName string `json:"shop_name"`
|
|
// PriceGranularity 说明价格是逐个 SKU 实测，还是只按颜色采样。
|
|
PriceGranularity string `json:"price_granularity"`
|
|
// Dimensions 决定规格各维度的显示顺序。
|
|
// 少了它就只能按 Go 的 map 遍历，而 map 是无序的——
|
|
// 同一个商品每次刷新页面「颜色/尺码」的先后都可能变。
|
|
Dimensions []struct {
|
|
Key string `json:"key"`
|
|
Name string `json:"name"`
|
|
} `json:"dimensions"`
|
|
SKUs []struct {
|
|
Options map[string]string `json:"options"`
|
|
PriceCent *int64 `json:"price_cent"` // 指针：采不到价格时是 null，不是 0
|
|
ListPriceCent *int64 `json:"list_price_cent"`
|
|
PriceObservedAt map[string]string `json:"price_observed_at"`
|
|
Available bool `json:"available"`
|
|
RawPrice string `json:"raw_price"`
|
|
} `json:"skus"`
|
|
}
|
|
|
|
// parseCollected 解析采集结果。解析不了直接报错，不要当成"采到 0 个规格"——
|
|
// 那是两回事：一个是客户端发的东西有问题，一个是商品确实没规格。
|
|
func parseCollected(raw string) (*collectedData, error) {
|
|
var c collectedData
|
|
if err := json.Unmarshal([]byte(raw), &c); err != nil {
|
|
return nil, fmt.Errorf("采集结果不是合法结构: %w", err)
|
|
}
|
|
if c.PriceGranularity != "" && c.PriceGranularity != "color" &&
|
|
c.PriceGranularity != "sku" {
|
|
return nil, fmt.Errorf(
|
|
"price_granularity 只能是 color 或 sku，收到 %q",
|
|
c.PriceGranularity)
|
|
}
|
|
return &c, nil
|
|
}
|
|
|
|
// artifactRef 从失败上报里取诊断产物位置。
|
|
//
|
|
// 按已定案的 Artifact 策略（docs/client/04 §10 待确认 #3），
|
|
// 客户端**只报本地引用、不上传文件**。所以这里存的是一个位置字符串，
|
|
// 告诉操作员去哪台客户端的哪个目录捞截图，Admin 自己显示不了图。
|
|
func artifactRef(req FailureRequest) string {
|
|
if len(req.Diagnostics) == 0 {
|
|
return ""
|
|
}
|
|
return string(req.Diagnostics)
|
|
}
|
|
|
|
// 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, info *repository.TaskInfo) (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
|
|
}
|
|
|
|
// 采集任务的结果还要落到 PDD 商品上，供后续规格匹配使用。
|
|
// 注意用的是 PddGoodsID —— 被采集的是 PDD 商品，不是蝦皮商品。
|
|
if info.TaskType == model.TaskCollect {
|
|
collected, err := parseCollected(pddData)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// 校验一：客户端采回来的商品必须就是我们要的那个。
|
|
// 链接跳转、采错商品时如果不拦，会把 B 的规格价格
|
|
// 存到 A 名下，后面按它下单就是买错东西。
|
|
if info.PddGoodsID != "" && collected.GoodsID != "" &&
|
|
collected.GoodsID != info.PddGoodsID {
|
|
return nil, fmt.Errorf(
|
|
"%w: 请求采集的是商品 %s，客户端返回的却是 %s",
|
|
ErrCollectMismatch, info.PddGoodsID, collected.GoodsID)
|
|
}
|
|
|
|
// 校验二：一个规格都没采到，对业务毫无用处
|
|
// （商品下架、页面改版、解析器没认出来）。
|
|
// 这种情况必须算失败，不能显示"已采集"让操作员空欢喜。
|
|
if len(collected.SKUs) == 0 {
|
|
if err := repository.SetCollectFailed(tx, info.PddGoodsID,
|
|
"未采集到任何规格，请检查商品是否已下架", ""); err != nil {
|
|
return nil, err
|
|
}
|
|
} else if err := repository.SetCollectResult(
|
|
tx, info.PddGoodsID, collected.Title, collected.ShopName, 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, info *repository.TaskInfo) (map[string]any, error) {
|
|
if err := repository.MarkTaskFailure(
|
|
tx, taskID, newStatus, req.Error.Code, req.Error.Message); err != nil {
|
|
return nil, err
|
|
}
|
|
// 采集失败要让操作员看得见原因和诊断产物位置
|
|
if info.TaskType == model.TaskCollect {
|
|
msg := req.Error.Message
|
|
if req.Error.Code != "" {
|
|
msg = req.Error.Code + ": " + msg
|
|
}
|
|
if err := repository.SetCollectFailed(
|
|
tx, info.PddGoodsID, msg, artifactRef(req)); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return map[string]any{
|
|
"accepted": true,
|
|
"result_id": newID(),
|
|
"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, info *repository.TaskInfo) (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. 任务得存在
|
|
info, err := repository.GetTaskInfo(tx, taskID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if info == nil {
|
|
return "", ErrTaskNotFound
|
|
}
|
|
|
|
// 3. 权限：**只有从没领过的才拒绝**。
|
|
// 任务已取消、已重派给别人，都照样接受。
|
|
claimed, err := repository.HasEverClaimed(tx, taskID, clientID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if !claimed {
|
|
return "", ErrNeverClaimed
|
|
}
|
|
|
|
// 4. 业务写入
|
|
resp, err := apply(tx, info)
|
|
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
|
|
}
|