feat: 任务改用采集采购独立序号 (#172)

2026-08-12 09:21:21 +08:00
parent eca3a77084
commit 9ad067503e
15 changed files with 523 additions and 44 deletions
@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"database/sql"
+	"errors"
 	"fmt"
 	"log"
 	"net"
@@ -19,7 +20,7 @@ import (
 	"cmautobuy/admin/spec"
 )

-const mysqlSchemaVersion = 12
+const mysqlSchemaVersion = 13

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

+// migrateMySQLV13 把任务真实主键迁为采集 cjN、采购 cgN 两套独立业务单号。
+//
+// 外键 DDL 可重放；主键和关联数据在单一事务中同时切换。若进程在记录版本前退出，
+// 再次启动会识别已经合规的编号，只校准序列，不会二次重编号。
+func migrateMySQLV13(db *sql.DB) error {
+	if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS task_sequences (
+		task_type VARCHAR(20) COLLATE utf8mb4_bin PRIMARY KEY,
+		current_value BIGINT NOT NULL DEFAULT 0,
+		CONSTRAINT chk_task_sequences_type CHECK (task_type IN ('collect','purchase')),
+		CONSTRAINT chk_task_sequences_value CHECK (current_value >= 0)
+	) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`); err != nil {
+		return fmt.Errorf("建立任务序列表失败: %w", err)
+	}
+	if _, err := db.Exec(`INSERT INTO task_sequences(task_type,current_value)
+		VALUES('collect',0),('purchase',0)
+		ON DUPLICATE KEY UPDATE task_type=VALUES(task_type)`); err != nil {
+		return fmt.Errorf("初始化任务序列失败: %w", err)
+	}
+	if err := ensureTaskForeignKeyCascade(db, "task_syb_sources", "fk_task_syb_sources_task"); err != nil {
+		return err
+	}
+	if err := migrateTaskIDsV13(db); err != nil {
+		return err
+	}
+	if err := ensureTaskForeignKeyCascade(db, "task_claims", "fk_task_claims_task"); err != nil {
+		return err
+	}
+	return nil
+}
+
+type taskIDMigrationV13 struct {
+	oldID, newID string
+	taskType     model.TaskType
+}
+
+func migrateTaskIDsV13(db *sql.DB) error {
+	tx, err := db.Begin()
+	if err != nil {
+		return fmt.Errorf("开始任务编号迁移事务失败: %w", err)
+	}
+	defer tx.Rollback()
+
+	rows, err := tx.Query(`SELECT task_id,task_type FROM tasks ORDER BY task_type,created_at,task_id`)
+	if err != nil {
+		return fmt.Errorf("读取待迁移任务失败: %w", err)
+	}
+	var source []taskIDMigrationV13
+	var hasLegacy, hasBusinessID bool
+	lastValues := map[model.TaskType]int64{model.TaskCollect: 0, model.TaskPurchase: 0}
+	for rows.Next() {
+		var taskID string
+		var taskType model.TaskType
+		if err := rows.Scan(&taskID, &taskType); err != nil {
+			rows.Close()
+			return fmt.Errorf("读取待迁移任务编号失败: %w", err)
+		}
+		if number, ok := parseTaskIDNumber(taskType, taskID); ok {
+			hasBusinessID = true
+			if number > lastValues[taskType] {
+				lastValues[taskType] = number
+			}
+		} else {
+			hasLegacy = true
+		}
+		source = append(source, taskIDMigrationV13{oldID: taskID, taskType: taskType})
+	}
+	if err := rows.Close(); err != nil {
+		return fmt.Errorf("关闭待迁移任务结果失败: %w", err)
+	}
+	if hasLegacy && hasBusinessID {
+		return fmt.Errorf("任务编号同时包含旧格式和 cj/cg 格式，拒绝自动混合迁移")
+	}
+	if _, err := tx.Exec(`DELETE c FROM task_claims c LEFT JOIN tasks t ON t.task_id=c.task_id WHERE t.task_id IS NULL`); err != nil {
+		return fmt.Errorf("清理孤儿任务领取历史失败: %w", err)
+	}
+	if hasLegacy {
+		if _, err := tx.Exec(`CREATE TEMPORARY TABLE task_id_map_v13 (
+			old_task_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY,
+			new_task_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL UNIQUE,
+			task_type VARCHAR(20) COLLATE utf8mb4_bin NOT NULL
+		) ENGINE=InnoDB`); err != nil {
+			return fmt.Errorf("建立任务编号临时映射失败: %w", err)
+		}
+		for i := range source {
+			lastValues[source[i].taskType]++
+			prefix, _ := taskIDPrefix(source[i].taskType)
+			source[i].newID = fmt.Sprintf("%s%d", prefix, lastValues[source[i].taskType])
+			if _, err := tx.Exec(`INSERT INTO task_id_map_v13(old_task_id,new_task_id,task_type) VALUES(?,?,?)`,
+				source[i].oldID, source[i].newID, source[i].taskType); err != nil {
+				return fmt.Errorf("保存任务编号映射 %s 失败: %w", source[i].oldID, err)
+			}
+		}
+		if _, err := tx.Exec(`UPDATE task_claims c JOIN task_id_map_v13 m ON m.old_task_id=c.task_id SET c.task_id=m.new_task_id`); err != nil {
+			return fmt.Errorf("迁移任务领取历史编号失败: %w", err)
+		}
+		if _, err := tx.Exec(`UPDATE tasks t JOIN task_id_map_v13 m ON m.old_task_id=t.task_id SET t.task_id=m.new_task_id`); err != nil {
+			return fmt.Errorf("迁移任务主键失败: %w", err)
+		}
+		if _, err := tx.Exec(`DROP TEMPORARY TABLE task_id_map_v13`); err != nil {
+			return fmt.Errorf("清理任务编号临时映射失败: %w", err)
+		}
+	}
+	for _, taskType := range []model.TaskType{model.TaskCollect, model.TaskPurchase} {
+		if _, err := tx.Exec(`UPDATE task_sequences SET current_value=? WHERE task_type=?`, lastValues[taskType], taskType); err != nil {
+			return fmt.Errorf("校准 %s 任务序列失败: %w", taskType, err)
+		}
+	}
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("提交任务编号迁移失败: %w", err)
+	}
+	return nil
+}
+
+func ensureTaskForeignKeyCascade(db *sql.DB, table, constraint string) error {
+	var deleteRule, updateRule string
+	err := db.QueryRow(`SELECT delete_rule,update_rule FROM information_schema.referential_constraints
+		WHERE constraint_schema=DATABASE() AND table_name=? AND constraint_name=?`, table, constraint).
+		Scan(&deleteRule, &updateRule)
+	if err == nil && deleteRule == "CASCADE" && updateRule == "CASCADE" {
+		return nil
+	}
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return fmt.Errorf("检查任务关联外键 %s 失败: %w", constraint, err)
+	}
+	if err == nil {
+		var dropStatement string
+		switch table {
+		case "task_syb_sources":
+			dropStatement = `ALTER TABLE task_syb_sources DROP FOREIGN KEY fk_task_syb_sources_task`
+		case "task_claims":
+			dropStatement = `ALTER TABLE task_claims DROP FOREIGN KEY fk_task_claims_task`
+		default:
+			return fmt.Errorf("不支持的任务关联表 %s", table)
+		}
+		if _, err := db.Exec(dropStatement); err != nil {
+			return fmt.Errorf("删除旧任务关联外键 %s 失败: %w", constraint, err)
+		}
+	}
+	var statement string
+	switch table {
+	case "task_syb_sources":
+		statement = `ALTER TABLE task_syb_sources ADD CONSTRAINT fk_task_syb_sources_task
+			FOREIGN KEY(task_id) REFERENCES tasks(task_id) ON DELETE CASCADE ON UPDATE CASCADE`
+	case "task_claims":
+		statement = `ALTER TABLE task_claims ADD CONSTRAINT fk_task_claims_task
+			FOREIGN KEY(task_id) REFERENCES tasks(task_id) ON DELETE CASCADE ON UPDATE CASCADE`
+	default:
+		return fmt.Errorf("不支持的任务关联表 %s", table)
+	}
+	if _, err := db.Exec(statement); err != nil {
+		return fmt.Errorf("建立任务关联外键 %s 失败: %w", constraint, err)
+	}
+	return nil
+}
+
+func checkMySQLV13Shape(db *sql.DB) error {
+	exists, err := mysqlTableExists(db, "task_sequences")
+	if err != nil || !exists {
+		return fmt.Errorf("任务序列表缺失")
+	}
+	if err := checkMySQLVarcharColumn(db, "task_sequences", "task_type", 20, false, "utf8mb4_bin", ""); err != nil {
+		return err
+	}
+	var valueType, valueNullable string
+	var valueDefault sql.NullString
+	if err := db.QueryRow(`SELECT data_type,is_nullable,column_default FROM information_schema.columns
+		WHERE table_schema=DATABASE() AND table_name='task_sequences' AND column_name='current_value'`).
+		Scan(&valueType, &valueNullable, &valueDefault); err != nil || valueType != "bigint" ||
+		valueNullable != "NO" || !valueDefault.Valid || valueDefault.String != "0" {
+		return fmt.Errorf("任务序列值列结构不正确")
+	}
+	for _, item := range []struct{ table, constraint string }{
+		{"task_claims", "fk_task_claims_task"},
+		{"task_syb_sources", "fk_task_syb_sources_task"},
+	} {
+		var deleteRule, updateRule string
+		if err := db.QueryRow(`SELECT delete_rule,update_rule FROM information_schema.referential_constraints
+			WHERE constraint_schema=DATABASE() AND table_name=? AND constraint_name=?`, item.table, item.constraint).
+			Scan(&deleteRule, &updateRule); err != nil || deleteRule != "CASCADE" || updateRule != "CASCADE" {
+			return fmt.Errorf("任务关联外键 %s 规则不正确", item.constraint)
+		}
+	}
+	var sequenceRows, invalidSequences int
+	if err := db.QueryRow(`SELECT COUNT(*),COALESCE(SUM(task_type NOT IN ('collect','purchase') OR current_value<0),0)
+		FROM task_sequences`).Scan(&sequenceRows, &invalidSequences); err != nil || sequenceRows != 2 || invalidSequences != 0 {
+		return fmt.Errorf("任务序列表必须且只能包含 collect/purchase 两条非负序列")
+	}
+	var invalidTasks, orphanClaims, orphanSources int
+	if err := db.QueryRow(`SELECT COUNT(*) FROM tasks WHERE
+		(task_type='collect' AND task_id NOT REGEXP '^cj[1-9][0-9]*$') OR
+		(task_type='purchase' AND task_id NOT REGEXP '^cg[1-9][0-9]*$')`).Scan(&invalidTasks); err != nil || invalidTasks != 0 {
+		return fmt.Errorf("存在不符合 cj/cg 规范的任务主键")
+	}
+	if err := db.QueryRow(`SELECT COUNT(*) FROM task_claims c LEFT JOIN tasks t ON t.task_id=c.task_id WHERE t.task_id IS NULL`).Scan(&orphanClaims); err != nil || orphanClaims != 0 {
+		return fmt.Errorf("存在孤儿任务领取历史")
+	}
+	if err := db.QueryRow(`SELECT COUNT(*) FROM task_syb_sources s LEFT JOIN tasks t ON t.task_id=s.task_id WHERE t.task_id IS NULL`).Scan(&orphanSources); err != nil || orphanSources != 0 {
+		return fmt.Errorf("存在孤儿顺运宝任务来源")
+	}
+	for _, taskType := range []model.TaskType{model.TaskCollect, model.TaskPurchase} {
+		prefix, _ := taskIDPrefix(taskType)
+		var lastValue, maxTaskNumber int64
+		if err := db.QueryRow(`SELECT current_value FROM task_sequences WHERE task_type=?`, taskType).Scan(&lastValue); err != nil {
+			return fmt.Errorf("读取 %s 任务序列失败: %w", taskType, err)
+		}
+		if err := db.QueryRow(`SELECT COALESCE(MAX(CAST(SUBSTRING(task_id,3) AS UNSIGNED)),0)
+			FROM tasks WHERE task_type=? AND task_id LIKE ?`, taskType, prefix+"%").Scan(&maxTaskNumber); err != nil {
+			return fmt.Errorf("读取 %s 最大任务编号失败: %w", taskType, err)
+		}
+		if lastValue < maxTaskNumber {
+			return fmt.Errorf("%s 任务序列 %d 小于现存最大编号 %d", taskType, lastValue, maxTaskNumber)
+		}
+	}
+	return nil
+}
+
 // migrateMySQLV12 保存采集任务的顺运宝来源，并回收没有有效任务的孤儿采集中状态。
 // CREATE TABLE IF NOT EXISTS 和带条件的 UPDATE 都可重放，适合 MySQL DDL 隐式提交后的重启恢复。
 func migrateMySQLV12(db *sql.DB) error {
@@ -1260,6 +1489,7 @@ func CheckMySQLSchema(db *sql.DB) error {
 		"spec_mapping_decisions",
 		"catalog_import_runs",
 		"task_syb_sources",
+		"task_sequences",
 	}
 	if err := checkMySQLSchema(db, mysqlRequiredTables); err != nil {
 		return err
@@ -1291,7 +1521,10 @@ func CheckMySQLSchema(db *sql.DB) error {
 	if err := checkMySQLV11Shape(db); err != nil {
 		return err
 	}
-	return checkMySQLV12Shape(db)
+	if err := checkMySQLV12Shape(db); err != nil {
+		return err
+	}
+	return checkMySQLV13Shape(db)
 }

 func checkMySQLV9Shape(db *sql.DB) error {