fix: 修复顺运宝采集任务孤儿状态 (#165)

2026-08-11 17:00:33 +08:00
parent c8b1fd686a
commit 9e328984e9
17 changed files with 508 additions and 67 deletions
@@ -19,7 +19,7 @@ import (
 	"cmautobuy/admin/spec"
 )

-const mysqlSchemaVersion = 11
+const mysqlSchemaVersion = 12

 // OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。
 func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) {
@@ -555,10 +555,102 @@ func MigrateMySQL(db *sql.DB) error {
 		if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 11, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
 			return fmt.Errorf("记录 MySQL schema v11 失败: %w", err)
 		}
+		current = 11
+	}
+	if current < 12 {
+		if err := migrateMySQLV12(db); err != nil {
+			return fmt.Errorf("执行 MySQL schema v12 失败: %w", err)
+		}
+		if err := checkMySQLV12Shape(db); err != nil {
+			return fmt.Errorf("MySQL schema v12 自检失败，未记录版本: %w", err)
+		}
+		if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 12, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
+			return fmt.Errorf("记录 MySQL schema v12 失败: %w", err)
+		}
 	}
 	return CheckMySQLSchema(db)
 }

+// migrateMySQLV12 保存采集任务的顺运宝来源，并回收没有有效任务的孤儿采集中状态。
+// CREATE TABLE IF NOT EXISTS 和带条件的 UPDATE 都可重放，适合 MySQL DDL 隐式提交后的重启恢复。
+func migrateMySQLV12(db *sql.DB) error {
+	if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS task_syb_sources (
+		task_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+		syb_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+		created_at VARCHAR(35) NOT NULL,
+		PRIMARY KEY (task_id, syb_id),
+		KEY idx_task_syb_sources_syb (syb_id, task_id),
+		CONSTRAINT fk_task_syb_sources_task FOREIGN KEY (task_id)
+			REFERENCES tasks(task_id) ON DELETE CASCADE,
+		CONSTRAINT fk_task_syb_sources_syb FOREIGN KEY (syb_id)
+			REFERENCES syb_orders(syb_id) ON DELETE CASCADE
+	) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`); err != nil {
+		return fmt.Errorf("建立采集任务顺运宝来源表失败: %w", err)
+	}
+	now := model.NowISO()
+	if _, err := db.Exec(`UPDATE pdd_products AS pp
+		SET collect_status='pending', collect_msg=NULL, updated_at=?
+		WHERE pp.collect_status='collecting' AND pp.deleted_at IS NULL
+		  AND NOT EXISTS (
+			SELECT 1 FROM tasks t
+			 WHERE t.task_type='collect' AND t.pdd_goods_id=pp.goods_id
+			   AND t.status IN ('pending','assigned','claimed')
+		  )`, now); err != nil {
+		return fmt.Errorf("回收无有效任务的 PDD 采集中状态失败: %w", err)
+	}
+	return nil
+}
+
+func checkMySQLV12Shape(db *sql.DB) error {
+	exists, err := mysqlTableExists(db, "task_syb_sources")
+	if err != nil || !exists {
+		return fmt.Errorf("采集任务顺运宝来源表缺失")
+	}
+	for _, column := range []struct {
+		name      string
+		length    int64
+		collation string
+	}{
+		{"task_id", 191, "utf8mb4_bin"},
+		{"syb_id", 191, "utf8mb4_bin"},
+		{"created_at", 35, "utf8mb4_0900_ai_ci"},
+	} {
+		if err := checkMySQLVarcharColumn(db, "task_syb_sources", column.name, column.length, false, column.collation, ""); err != nil {
+			return err
+		}
+	}
+	for _, index := range []struct{ name, columns string }{
+		{"PRIMARY", "task_id,syb_id"},
+		{"idx_task_syb_sources_syb", "syb_id,task_id"},
+	} {
+		var columns string
+		if err := db.QueryRow(`SELECT GROUP_CONCAT(column_name ORDER BY seq_in_index)
+			FROM information_schema.statistics
+			WHERE table_schema=DATABASE() AND table_name='task_syb_sources' AND index_name=?`, index.name).Scan(&columns); err != nil || columns != index.columns {
+			return fmt.Errorf("采集任务顺运宝来源索引 %s 不正确", index.name)
+		}
+	}
+	for _, foreignKey := range []struct{ name, column, table, target string }{
+		{"fk_task_syb_sources_task", "task_id", "tasks", "task_id"},
+		{"fk_task_syb_sources_syb", "syb_id", "syb_orders", "syb_id"},
+	} {
+		var column, table, target string
+		err := db.QueryRow(`SELECT column_name,referenced_table_name,referenced_column_name
+			FROM information_schema.key_column_usage
+			WHERE constraint_schema=DATABASE() AND table_name='task_syb_sources' AND constraint_name=?`, foreignKey.name).
+			Scan(&column, &table, &target)
+		if err != nil || column != foreignKey.column || table != foreignKey.table || target != foreignKey.target {
+			return fmt.Errorf("采集任务顺运宝来源约束 %s 不正确", foreignKey.name)
+		}
+		var deleteRule string
+		if err := db.QueryRow(`SELECT delete_rule FROM information_schema.referential_constraints
+			WHERE constraint_schema=DATABASE() AND table_name='task_syb_sources' AND constraint_name=?`, foreignKey.name).Scan(&deleteRule); err != nil || deleteRule != "CASCADE" {
+			return fmt.Errorf("采集任务顺运宝来源约束 %s 删除规则不正确", foreignKey.name)
+		}
+	}
+	return nil
+}
+
 // migrateMySQLV11 增加顺运宝店铺列，并从保留的原始 JSON 回填历史数据。
 // DDL 和 UPDATE 都可重放：中断后再次启动只补缺列和仍为空的记录。
 func migrateMySQLV11(db *sql.DB) error {
@@ -1167,6 +1259,7 @@ func CheckMySQLSchema(db *sql.DB) error {
 		"users", "web_sessions", "client_user_assignments", "syb_sync_runs", "admin_initialization_lock",
 		"spec_mapping_decisions",
 		"catalog_import_runs",
+		"task_syb_sources",
 	}
 	if err := checkMySQLSchema(db, mysqlRequiredTables); err != nil {
 		return err
@@ -1189,7 +1282,16 @@ func CheckMySQLSchema(db *sql.DB) error {
 	if err := checkMySQLV8Shape(db); err != nil {
 		return err
 	}
-	return checkMySQLV9Shape(db)
+	if err := checkMySQLV9Shape(db); err != nil {
+		return err
+	}
+	if err := checkMySQLV10Shape(db); err != nil {
+		return err
+	}
+	if err := checkMySQLV11Shape(db); err != nil {
+		return err
+	}
+	return checkMySQLV12Shape(db)
 }

 func checkMySQLV9Shape(db *sql.DB) error {