cmautobuy/admin/service/submit.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"`
	SKUs    []struct {
		Options   map[string]string `json:"options"`
		PriceCent *int64            `json:"price_cent"` // 指针：采不到价格时是 null，不是 0
		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)
	}
	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, 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,
				"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
}