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
)

// 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
}