feat: 增加真实采购任务安全模式 (#98)

admin/repository/client.go

@@ -2,6 +2,7 @@ package repository
 
 import (
 	"database/sql"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"strings"
@@ -9,6 +10,29 @@ import (
 	"cmautobuy/admin/model"
 )
 
+// ClientPurchaseMode 返回客户端最后一次登记时声明的采购能力。
+// 旧数据、空值或无法解析的 JSON 一律按 dry_run 处理，这是安全默认值。
+func ClientPurchaseMode(q Execer, clientID string) (string, error) {
+	var raw sql.NullString
+	if err := q.QueryRow(`SELECT capabilities FROM clients WHERE client_id = ?`, clientID).Scan(&raw); errors.Is(err, sql.ErrNoRows) {
+		return "", ErrClientNotFound
+	} else if err != nil {
+		return "", fmt.Errorf("读取客户端 %s 的采购能力失败: %w", clientID, err)
+	}
+	return ParseClientPurchaseMode(raw.String), nil
+}
+
+// ParseClientPurchaseMode 把登记能力转成安全的固定值，供列表和创建校验共用。
+func ParseClientPurchaseMode(raw string) string {
+	var capabilities struct {
+		PurchaseMode string `json:"purchase_mode"`
+	}
+	if json.Unmarshal([]byte(raw), &capabilities) == nil && capabilities.PurchaseMode == string(model.TaskExecutionLive) {
+		return capabilities.PurchaseMode
+	}
+	return string(model.TaskExecutionDryRun)
+}
+
 var (
 	ErrClientNotFound     = errors.New("客户端不存在")
 	ErrPurchaserNotActive = errors.New("采购员不存在或已禁用")

admin/repository/mysql_db.go

@@ -19,7 +19,7 @@ import (
 	"cmautobuy/admin/spec"
 )
 
-const mysqlSchemaVersion = 4
+const mysqlSchemaVersion = 5
 
 // OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。
 func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) {
@@ -471,10 +471,77 @@ func MigrateMySQL(db *sql.DB) error {
 		if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 4, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
 			return fmt.Errorf("记录 MySQL schema v4 失败: %w", err)
 		}
+		current = 4
+	}
+	if current < 5 {
+		if err := migrateMySQLV5(db); err != nil {
+			return fmt.Errorf("执行 MySQL schema v5 失败: %w", err)
+		}
+		if err := checkMySQLV5Shape(db); err != nil {
+			return fmt.Errorf("MySQL schema v5 自检失败，未记录版本: %w", err)
+		}
+		if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 5, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
+			return fmt.Errorf("记录 MySQL schema v5 失败: %w", err)
+		}
 	}
 	return CheckMySQLSchema(db)
 }
 
+// migrateMySQLV5 给任务增加不可变执行模式和真实下单创建审计。
+// 每条 DDL 都先检查存在性，MySQL 在任意一步隐式提交后都可以安全重放。
+func migrateMySQLV5(db *sql.DB) error {
+	columns := []struct {
+		name string
+		ddl  string
+	}{
+		{"execution_mode", `ALTER TABLE tasks ADD COLUMN execution_mode VARCHAR(16) COLLATE utf8mb4_bin NOT NULL DEFAULT 'dry_run' AFTER status`},
+		{"live_confirmed_by", `ALTER TABLE tasks ADD COLUMN live_confirmed_by VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER finished_at`},
+		{"live_confirmed_at", `ALTER TABLE tasks ADD COLUMN live_confirmed_at VARCHAR(35) NULL AFTER live_confirmed_by`},
+	}
+	for _, column := range columns {
+		exists, err := mysqlColumnExists(db, "tasks", column.name)
+		if err != nil {
+			return err
+		}
+		if !exists {
+			if _, err := db.Exec(column.ddl); err != nil {
+				return fmt.Errorf("增加 tasks.%s 失败: %w", column.name, err)
+			}
+		}
+	}
+	if _, err := db.Exec(`UPDATE tasks SET execution_mode='dry_run' WHERE execution_mode IS NULL OR execution_mode=''`); err != nil {
+		return fmt.Errorf("回填任务执行模式失败: %w", err)
+	}
+	constraints := []struct {
+		name string
+		ddl  string
+	}{
+		{"chk_tasks_execution_mode", `ALTER TABLE tasks ADD CONSTRAINT chk_tasks_execution_mode CHECK (execution_mode IN ('dry_run','live'))`},
+		{"chk_tasks_live_confirmation", `ALTER TABLE tasks ADD CONSTRAINT chk_tasks_live_confirmation CHECK ((execution_mode='dry_run' AND live_confirmed_by IS NULL AND live_confirmed_at IS NULL) OR (execution_mode='live' AND live_confirmed_by IS NOT NULL AND live_confirmed_by<>'' AND live_confirmed_at IS NOT NULL AND live_confirmed_at<>''))`},
+	}
+	for _, constraint := range constraints {
+		exists, err := mysqlConstraintExists(db, "tasks", constraint.name)
+		if err != nil {
+			return err
+		}
+		if !exists {
+			if _, err := db.Exec(constraint.ddl); err != nil {
+				return fmt.Errorf("增加 tasks.%s 失败: %w", constraint.name, err)
+			}
+		}
+	}
+	indexExists, err := mysqlIndexExists(db, "tasks", "idx_tasks_claim_mode")
+	if err != nil {
+		return err
+	}
+	if !indexExists {
+		if _, err := db.Exec(`ALTER TABLE tasks ADD INDEX idx_tasks_claim_mode (assigned_client, status, execution_mode, priority DESC, created_at)`); err != nil {
+			return fmt.Errorf("增加任务模式领取索引失败: %w", err)
+		}
+	}
+	return nil
+}
+
 func checkMySQLSchemaV3(db *sql.DB) error {
 	tables := []string{"shopee_products", "shopee_skus", "pdd_products", "syb_orders", "spec_mappings", "tasks", "clients", "idempotency_keys", "task_claims", "syb_session", "syb_sync_state", "users", "web_sessions", "client_user_assignments", "syb_sync_runs", "admin_initialization_lock"}
 	if err := checkMySQLSchema(db, tables); err != nil {
@@ -563,6 +630,15 @@ func mysqlConstraintExists(db *sql.DB, table, constraint string) (bool, error) {
 	return count == 1, nil
 }
 
+func mysqlIndexExists(db *sql.DB, table, index string) (bool, error) {
+	var count int
+	if err := db.QueryRow(`SELECT COUNT(DISTINCT index_name) FROM information_schema.statistics
+		WHERE table_schema=DATABASE() AND table_name=? AND index_name=?`, table, index).Scan(&count); err != nil {
+		return false, fmt.Errorf("检查 MySQL 索引 %s.%s 失败: %w", table, index, err)
+	}
+	return count == 1, nil
+}
+
 func backfillSybSpecKeys(db *sql.DB) error {
 	const batchSize = 500
 	cursor := ""
@@ -729,7 +805,65 @@ func CheckMySQLSchema(db *sql.DB) error {
 	if err := checkMySQLV3Shape(db); err != nil {
 		return err
 	}
-	return checkMySQLV4Shape(db)
+	if err := checkMySQLV4Shape(db); err != nil {
+		return err
+	}
+	return checkMySQLV5Shape(db)
+}
+
+func checkMySQLV5Shape(db *sql.DB) error {
+	if err := checkMySQLVarcharColumn(db, "tasks", "execution_mode", 16, false, "utf8mb4_bin", "dry_run"); err != nil {
+		return err
+	}
+	if err := checkMySQLVarcharColumn(db, "tasks", "live_confirmed_by", 191, true, "utf8mb4_bin", ""); err != nil {
+		return err
+	}
+	if err := checkMySQLNullDefault(db, "tasks", "live_confirmed_by"); err != nil {
+		return err
+	}
+	if err := checkMySQLVarcharColumn(db, "tasks", "live_confirmed_at", 35, true, "utf8mb4_0900_ai_ci", ""); err != nil {
+		return err
+	}
+	if err := checkMySQLNullDefault(db, "tasks", "live_confirmed_at"); err != nil {
+		return err
+	}
+	for _, check := range []struct {
+		name     string
+		required []string
+	}{
+		{"chk_tasks_execution_mode", []string{"execution_modein'dry_run','live'"}},
+		{"chk_tasks_live_confirmation", []string{"execution_mode='dry_run'", "live_confirmed_byisnull", "live_confirmed_atisnull", "execution_mode='live'", "live_confirmed_byisnotnull", "live_confirmed_by<>''", "live_confirmed_atisnotnull", "live_confirmed_at<>''"}},
+	} {
+		var enforced, clause string
+		if err := db.QueryRow(`SELECT tc.enforced,cc.check_clause FROM information_schema.table_constraints tc JOIN information_schema.check_constraints cc ON cc.constraint_schema=tc.constraint_schema AND cc.constraint_name=tc.constraint_name WHERE tc.constraint_schema=DATABASE() AND tc.table_name='tasks' AND tc.constraint_name=? AND tc.constraint_type='CHECK'`, check.name).Scan(&enforced, &clause); err != nil {
+			return fmt.Errorf("任务 CHECK %s 缺失或不可读: %w", check.name, err)
+		}
+		normalized := strings.NewReplacer("`", "", " ", "", "(", "", ")", "", "_utf8mb4", "", `\`, "").Replace(strings.ToLower(clause))
+		if enforced != "YES" {
+			return fmt.Errorf("任务 CHECK %s 未启用", check.name)
+		}
+		for _, fragment := range check.required {
+			if !strings.Contains(normalized, fragment) {
+				return fmt.Errorf("任务 CHECK %s 表达式不正确", check.name)
+			}
+		}
+	}
+	var cols string
+	if err := db.QueryRow(`SELECT GROUP_CONCAT(CONCAT(column_name,':',collation) ORDER BY seq_in_index) FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='tasks' AND index_name='idx_tasks_claim_mode'`).Scan(&cols); err != nil || cols != "assigned_client:A,status:A,execution_mode:A,priority:D,created_at:A" {
+		return fmt.Errorf("任务模式领取索引不正确")
+	}
+	return nil
+}
+
+func checkMySQLNullDefault(db *sql.DB, table, column string) error {
+	var defaultValue sql.NullString
+	if err := db.QueryRow(`SELECT column_default FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name=? AND column_name=?`, table, column).Scan(&defaultValue); err != nil {
+		return fmt.Errorf("检查 MySQL 列 %s.%s 默认值失败: %w", table, column, err)
+	}
+	if defaultValue.Valid {
+		return fmt.Errorf("MySQL 列 %s.%s 默认值应为 NULL", table, column)
+	}
+	return nil
 }
 
 func checkMySQLV4Shape(db *sql.DB) error {

admin/repository/mysql_db_integration_test.go

@@ -232,6 +232,60 @@ func TestMySQLMigrate_V4形状错误不记版本(t *testing.T) {
 	}
 }
 
+func TestMySQLMigrate_V4升级V5且断点重跑(t *testing.T) {
+	db := openMySQLMigrationTestDB(t)
+	defer db.Close()
+	cleanMySQLTestSchema(t, db)
+	defer cleanMySQLTestSchema(t, db)
+	prepareMySQLV4(t, db)
+	// 模拟全部 DDL 已完成但版本尚未记录。
+	if err := migrateMySQLV5(db); err != nil {
+		t.Fatal(err)
+	}
+	if err := MigrateMySQL(db); err != nil {
+		t.Fatal(err)
+	}
+	if err := MigrateMySQL(db); err != nil {
+		t.Fatalf("v5 重跑失败: %v", err)
+	}
+	var versions int
+	if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=5`).Scan(&versions); err != nil || versions != 1 {
+		t.Fatalf("v5=%d err=%v", versions, err)
+	}
+	now := "2026-08-10T00:00:00Z"
+	mustExec(t, db, `INSERT INTO tasks(task_id,task_type,status,pdd_goods_url,created_at,updated_at) VALUES('OLD','collect','pending','https://example.invalid',?,?)`, now, now)
+	var mode string
+	if err := db.QueryRow(`SELECT execution_mode FROM tasks WHERE task_id='OLD'`).Scan(&mode); err != nil || mode != "dry_run" {
+		t.Fatalf("历史/缺省任务模式=%q err=%v", mode, err)
+	}
+	if _, err := db.Exec(`UPDATE tasks SET execution_mode='invalid' WHERE task_id='OLD'`); err == nil {
+		t.Fatal("执行模式 CHECK 必须拒绝非法值")
+	}
+	if _, err := db.Exec(`UPDATE tasks SET execution_mode='live' WHERE task_id='OLD'`); err == nil {
+		t.Fatal("缺少确认审计的 live 必须被拒绝")
+	}
+	if _, err := db.Exec(`UPDATE tasks SET live_confirmed_by='U',live_confirmed_at=? WHERE task_id='OLD'`, now); err == nil {
+		t.Fatal("dry_run 不得携带 live 确认审计")
+	}
+}
+
+func TestMySQLMigrate_V5形状错误不记版本(t *testing.T) {
+	db := openMySQLMigrationTestDB(t)
+	defer db.Close()
+	cleanMySQLTestSchema(t, db)
+	defer cleanMySQLTestSchema(t, db)
+	prepareMySQLV4(t, db)
+	mustExec(t, db, `ALTER TABLE tasks ADD COLUMN execution_mode VARCHAR(8) NOT NULL DEFAULT 'dry_run'`)
+	if err := MigrateMySQL(db); err == nil {
+		t.Fatal("错误 execution_mode 形状必须阻止 v5")
+	}
+	var count int
+	db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=5`).Scan(&count)
+	if count != 0 {
+		t.Fatal("v5 自检失败不得记录版本")
+	}
+}
+
 func openMySQLMigrationTestDB(t *testing.T) *sql.DB {
 	t.Helper()
 	if os.Getenv("CMAUTOBUY_MYSQL_TEST") != "1" {
@@ -263,6 +317,16 @@ func prepareMySQLV2(t *testing.T, db *sql.DB) {
 	mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (1,'2026-08-10T00:00:00Z'),(2,'2026-08-10T00:00:00Z')`)
 }
 
+func prepareMySQLV4(t *testing.T, db *sql.DB) {
+	t.Helper()
+	prepareMySQLV2(t, db)
+	if err := migrateMySQLV3(db); err != nil {
+		t.Fatal(err)
+	}
+	mustExec(t, db, mysqlSchemaV4Decisions)
+	mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (3,'2026-08-10T00:00:00Z'),(4,'2026-08-10T00:00:00Z')`)
+}
+
 func mustExec(t *testing.T, db *sql.DB, query string, args ...any) {
 	t.Helper()
 	if _, err := db.Exec(query, args...); err != nil {

admin/repository/task.go

@@ -30,10 +30,17 @@ import (
 //
 // InnoDB 事务用 FOR UPDATE SKIP LOCKED 锁住一条候选任务。领取状态和
 // task_claims 历史在同一个事务提交，避免只改了状态却没留下领取凭据。
-func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model.Task, error) {
+func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string, purchaseModes ...string) (*model.Task, error) {
 	if clientID == "" {
 		return nil, fmt.Errorf("client_id 不能为空")
 	}
+	purchaseMode := string(model.TaskExecutionDryRun)
+	if len(purchaseModes) > 0 {
+		purchaseMode = purchaseModes[0]
+	}
+	if purchaseMode != string(model.TaskExecutionDryRun) && purchaseMode != string(model.TaskExecutionLive) {
+		return nil, fmt.Errorf("purchase_mode 必须是 dry_run 或 live")
+	}
 	ctx := context.Background()
 	tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
 	if err != nil {
@@ -45,6 +52,11 @@ func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model
 	          WHERE (   (assigned_client = ?    AND status = 'assigned')
 	                 OR (assigned_client IS NULL AND status = 'pending') )`
 	args := []any{clientID}
+	// dry_run 客户端永远看不到真实采购任务。声明 live 的客户端仍可执行
+	// 演练任务，避免它在没有真实任务时闲置。
+	if purchaseMode != string(model.TaskExecutionLive) {
+		query += ` AND execution_mode = 'dry_run'`
+	}
 
 	// 客户端只声明支持某些类型时，不要给它别的类型
 	if len(supportedTypes) > 0 {
@@ -101,23 +113,26 @@ func GetTask(db *sql.DB, taskID string) (*model.Task, error) {
 	var t model.Task
 	var assigned, claimedAt, sybID, orderNo, goodsID, skuID sql.NullString
 	var pddGoodsID, pddOptions, resultData, errCode, errMsg, finishedAt sql.NullString
+	var liveConfirmedBy, liveConfirmedAt sql.NullString
 	var quantity, maxPrice sql.NullInt64
 
 	err := db.QueryRow(`
-		SELECT task_id, task_type, status, version, priority,
+		SELECT task_id, task_type, status, execution_mode, version, priority,
 		       assigned_client, claimed_at,
 		       syb_id, order_no, goods_id, shopee_sku_id,
 		       pdd_goods_url, pdd_goods_id, pdd_options,
 		       quantity, max_price_cent,
 		       result_data, error_code, error_message, finished_at,
+		       live_confirmed_by, live_confirmed_at,
 		       created_at, updated_at
 		FROM tasks WHERE task_id = ?`, taskID).Scan(
-		&t.TaskID, &t.TaskType, &t.Status, &t.Version, &t.Priority,
+		&t.TaskID, &t.TaskType, &t.Status, &t.ExecutionMode, &t.Version, &t.Priority,
 		&assigned, &claimedAt,
 		&sybID, &orderNo, &goodsID, &skuID,
 		&t.PddGoodsURL, &pddGoodsID, &pddOptions,
 		&quantity, &maxPrice,
 		&resultData, &errCode, &errMsg, &finishedAt,
+		&liveConfirmedBy, &liveConfirmedAt,
 		&t.CreatedAt, &t.UpdatedAt)
 	if err == sql.ErrNoRows {
 		return nil, nil
@@ -140,6 +155,8 @@ func GetTask(db *sql.DB, taskID string) (*model.Task, error) {
 	t.ErrorCode = errCode.String
 	t.ErrorMessage = errMsg.String
 	t.FinishedAt = finishedAt.String
+	t.LiveConfirmedBy = liveConfirmedBy.String
+	t.LiveConfirmedAt = liveConfirmedAt.String
 	return &t, nil
 }
 
@@ -269,6 +286,7 @@ type TaskListRow struct {
 	TaskID         string
 	TaskType       model.TaskType
 	Status         model.TaskStatus
+	ExecutionMode  model.TaskExecutionMode
 	AssignedClient string // 空表示无主任务
 
 	OrderNo      string
@@ -323,7 +341,7 @@ func taskFilterClause(filter TaskFilter) (string, []any) {
 func ListTasks(q Execer, filter TaskFilter, limit, offset int) ([]TaskListRow, error) {
 	where, args := taskFilterClause(filter)
 	sqlText := `
-		SELECT t.task_id, t.task_type, t.status, t.assigned_client,
+		SELECT t.task_id, t.task_type, t.status, t.execution_mode, t.assigned_client,
 		       t.order_no, t.pdd_goods_id, t.pdd_options, t.quantity, t.max_price_cent,
 		       t.updated_at, p.title
 		  FROM tasks t
@@ -345,7 +363,7 @@ func ListTasks(q Execer, filter TaskFilter, limit, offset int) ([]TaskListRow, e
 		var quantity, maxPrice sql.NullInt64
 
 		if err := rows.Scan(
-			&r.TaskID, &r.TaskType, &r.Status, &assigned,
+			&r.TaskID, &r.TaskType, &r.Status, &r.ExecutionMode, &assigned,
 			&orderNo, &pddGoodsID, &pddOptions, &quantity, &maxPrice,
 			&r.UpdatedAt, &title,
 		); err != nil {
@@ -465,17 +483,34 @@ func InsertPurchaseTask(q Execer, task model.Task) error {
 		task.Quantity <= 0 || task.MaxPriceCent <= 0 {
 		return fmt.Errorf("采购任务缺少客户端、商品、规格、数量或人民币价格上限")
 	}
+	if task.ExecutionMode == "" {
+		task.ExecutionMode = model.TaskExecutionDryRun
+	}
+	if task.ExecutionMode != model.TaskExecutionDryRun && task.ExecutionMode != model.TaskExecutionLive {
+		return fmt.Errorf("采购任务执行模式无效")
+	}
+	if task.ExecutionMode == model.TaskExecutionLive &&
+		(strings.TrimSpace(task.LiveConfirmedBy) == "" || strings.TrimSpace(task.LiveConfirmedAt) == "") {
+		return fmt.Errorf("真实采购任务缺少创建确认审计信息")
+	}
+	if task.ExecutionMode == model.TaskExecutionDryRun {
+		task.LiveConfirmedBy = ""
+		task.LiveConfirmedAt = ""
+	}
 	now := model.NowISO()
+	liveConfirmedBy := sql.NullString{String: task.LiveConfirmedBy, Valid: task.LiveConfirmedBy != ""}
+	liveConfirmedAt := sql.NullString{String: task.LiveConfirmedAt, Valid: task.LiveConfirmedAt != ""}
 	_, err := q.Exec(`
 		INSERT INTO tasks
-		       (task_id, task_type, status, assigned_client,
+		       (task_id, task_type, status, execution_mode, assigned_client,
 		        syb_id, order_no, goods_id, shopee_sku_id,
 		        pdd_goods_url, pdd_goods_id, pdd_options,
-		        quantity, max_price_cent, created_at, updated_at)
-		VALUES (?, 'purchase', 'assigned', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
-		task.TaskID, task.AssignedClient, task.SybID, task.OrderNo,
+		        quantity, max_price_cent, live_confirmed_by, live_confirmed_at,
+		        created_at, updated_at)
+		VALUES (?, 'purchase', 'assigned', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		task.TaskID, task.ExecutionMode, task.AssignedClient, task.SybID, task.OrderNo,
 		task.GoodsID, task.ShopeeSKUID, task.PddGoodsURL, task.PddGoodsID,
-		task.PddOptions, task.Quantity, task.MaxPriceCent, now, now)
+		task.PddOptions, task.Quantity, task.MaxPriceCent, liveConfirmedBy, liveConfirmedAt, now, now)
 	if err != nil {
 		return fmt.Errorf("创建顺运宝明细 %s 的采购任务失败: %w", task.SybID, err)
 	}