From 9ad067503eaed100939c4d4e2820129e24ea9d2b Mon Sep 17 00:00:00 2001 From: chengma Date: Wed, 12 Aug 2026 09:21:21 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BB=BB=E5=8A=A1=E6=94=B9=E7=94=A8?= =?UTF-8?q?=E9=87=87=E9=9B=86=E9=87=87=E8=B4=AD=E7=8B=AC=E7=AB=8B=E5=BA=8F?= =?UTF-8?q?=E5=8F=B7=20(#172)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- admin/repository/mysql_db.go | 237 +++++++++++++++++- admin/repository/mysql_db_integration_test.go | 149 +++++++++++ admin/repository/task.go | 60 +++++ admin/service/pdd.go | 21 +- admin/service/pdd_page_test.go | 4 + admin/service/purchase_workflow.go | 14 +- admin/service/purchase_workflow_test.go | 8 +- client/test/test_admin_gateway_contract.py | 8 +- client/test/test_http_admin_gateway.py | 4 +- docs/admin/01-requirements.md | 4 + docs/admin/03-data-model.md | 34 ++- docs/admin/04-client-api.md | 6 +- docs/admin/05-ui-specification.md | 10 +- docs/client/04-admin-api-contract.md | 6 +- docs/client/05-ui-specification.md | 2 +- 15 files changed, 523 insertions(+), 44 deletions(-) diff --git a/admin/repository/mysql_db.go b/admin/repository/mysql_db.go index 40826b1..92797e0 100644 --- a/admin/repository/mysql_db.go +++ b/admin/repository/mysql_db.go @@ -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 { diff --git a/admin/repository/mysql_db_integration_test.go b/admin/repository/mysql_db_integration_test.go index e90cd1d..8e5a270 100644 --- a/admin/repository/mysql_db_integration_test.go +++ b/admin/repository/mysql_db_integration_test.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "strings" + "sync" "testing" "cmautobuy/admin/config" @@ -650,6 +651,154 @@ func TestMySQLMigrate_V12形状错误不记录版本(t *testing.T) { } } +func TestMySQLMigrate_V12升级V13迁移任务主键和关联(t *testing.T) { + db := openMySQLMigrationTestDB(t) + defer db.Close() + cleanMySQLTestSchema(t, db) + defer cleanMySQLTestSchema(t, db) + if err := MigrateMySQL(db); err != nil { + t.Fatal(err) + } + // 模拟生产 v12：移除 v13 表、版本和新增的级联更新外键。 + mustExec(t, db, `DELETE FROM schema_migrations WHERE version=13`) + mustExec(t, db, `DROP TABLE task_sequences`) + mustExec(t, db, `ALTER TABLE task_claims DROP FOREIGN KEY fk_task_claims_task`) + mustExec(t, db, `ALTER TABLE task_syb_sources DROP FOREIGN KEY fk_task_syb_sources_task`) + mustExec(t, db, `ALTER TABLE task_syb_sources ADD CONSTRAINT fk_task_syb_sources_task + FOREIGN KEY(task_id) REFERENCES tasks(task_id) ON DELETE CASCADE`) + + mustExec(t, db, `INSERT INTO syb_orders(syb_id,order_no,title,product_spec,quantity,syb_data,created_at,updated_at) + VALUES('SYB-V13','ORDER-SOURCE','商品','黑色,M',1,'{}','2026-08-12T00:00:00Z','2026-08-12T00:00:00Z')`) + mustExec(t, db, `INSERT INTO tasks(task_id,task_type,status,pdd_goods_url,pdd_goods_id,order_no,result_data,created_at,updated_at) VALUES + ('OLD-C-LATE','collect','succeeded','https://example.invalid/c2','C-LATE',NULL,NULL,'2026-08-12T02:00:00Z','2026-08-12T02:00:00Z'), + ('OLD-C-FIRST','collect','assigned','https://example.invalid/c1','C-FIRST',NULL,NULL,'2026-08-12T01:00:00Z','2026-08-12T01:00:00Z'), + ('OLD-P-LATE','purchase','succeeded','https://example.invalid/p2','P-LATE','ORDER-P-LATE','{\"purchase\":{\"order_no\":\"PDD-2\",\"ordered_at\":\"2026-08-12T02:00:00Z\"}}','2026-08-12T04:00:00Z','2026-08-12T04:00:00Z'), + ('OLD-P-FIRST','purchase','succeeded','https://example.invalid/p1','P-FIRST','ORDER-P-FIRST','{\"purchase\":{\"order_no\":\"PDD-1\",\"ordered_at\":\"2026-08-12T01:00:00Z\"}}','2026-08-12T03:00:00Z','2026-08-12T03:00:00Z')`) + mustExec(t, db, `INSERT INTO task_claims(task_id,client_id,claimed_at) VALUES + ('OLD-C-FIRST','CLIENT-1','2026-08-12T01:10:00Z'), + ('OLD-P-FIRST','CLIENT-1','2026-08-12T03:10:00Z'), + ('DELETED-TASK','CLIENT-OLD','2026-08-01T00:00:00Z')`) + mustExec(t, db, `INSERT INTO task_syb_sources(task_id,syb_id,created_at) + VALUES('OLD-C-FIRST','SYB-V13','2026-08-12T01:00:00Z')`) + + if err := MigrateMySQL(db); err != nil { + t.Fatalf("v12 升级 v13 失败: %v", err) + } + wants := map[string]string{"C-FIRST": "cj1", "C-LATE": "cj2", "P-FIRST": "cg1", "P-LATE": "cg2"} + for goodsID, wantID := range wants { + var taskID string + if err := db.QueryRow(`SELECT task_id FROM tasks WHERE pdd_goods_id=?`, goodsID).Scan(&taskID); err != nil || taskID != wantID { + t.Fatalf("商品 %s 任务编号=%q err=%v，期望 %s", goodsID, taskID, err, wantID) + } + } + var claimCount, orphanCount, sourceCount int + if err := db.QueryRow(`SELECT COUNT(*) FROM task_claims WHERE task_id IN ('cj1','cg1')`).Scan(&claimCount); err != nil || claimCount != 2 { + t.Fatalf("有效领取历史未迁移: count=%d err=%v", claimCount, err) + } + 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(&orphanCount); err != nil || orphanCount != 0 { + t.Fatalf("孤儿领取历史未清理: count=%d err=%v", orphanCount, err) + } + if err := db.QueryRow(`SELECT COUNT(*) FROM task_syb_sources WHERE task_id='cj1' AND syb_id='SYB-V13'`).Scan(&sourceCount); err != nil || sourceCount != 1 { + t.Fatalf("顺运宝来源未迁移: count=%d err=%v", sourceCount, err) + } + var resultData string + if err := db.QueryRow(`SELECT result_data FROM tasks WHERE task_id='cg1'`).Scan(&resultData); err != nil || !strings.Contains(resultData, `"order_no":"PDD-1"`) { + t.Fatalf("采购结果不得变化: result=%q err=%v", resultData, err) + } + for taskType, want := range map[model.TaskType]int64{model.TaskCollect: 2, model.TaskPurchase: 2} { + var last int64 + if err := db.QueryRow(`SELECT current_value FROM task_sequences WHERE task_type=?`, taskType).Scan(&last); err != nil || last != want { + t.Fatalf("%s 序列=%d err=%v，期望 %d", taskType, last, err, want) + } + } + if err := MigrateMySQL(db); err != nil { + t.Fatalf("v13 重放失败: %v", err) + } + var stableID string + if err := db.QueryRow(`SELECT task_id FROM tasks WHERE pdd_goods_id='C-FIRST'`).Scan(&stableID); err != nil || stableID != "cj1" { + t.Fatalf("重放不得二次改号: id=%s err=%v", stableID, err) + } +} + +func TestNextTaskID_MySQL并发独立递增且回滚不耗号(t *testing.T) { + db := openMySQLMigrationTestDB(t) + defer db.Close() + cleanMySQLTestSchema(t, db) + defer cleanMySQLTestSchema(t, db) + if err := MigrateMySQL(db); err != nil { + t.Fatal(err) + } + + const workers = 16 + ids := make(map[string]bool, workers) + var mu sync.Mutex + errCh := make(chan error, workers) + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + tx, err := db.Begin() + if err != nil { + errCh <- err + return + } + id, err := NextTaskID(tx, model.TaskCollect) + if err == nil { + err = tx.Commit() + } else { + tx.Rollback() + } + if err != nil { + errCh <- err + return + } + mu.Lock() + ids[id] = true + mu.Unlock() + }() + } + wg.Wait() + close(errCh) + for err := range errCh { + t.Fatal(err) + } + if len(ids) != workers { + t.Fatalf("并发分配得到 %d 个唯一编号，期望 %d", len(ids), workers) + } + for i := 1; i <= workers; i++ { + if !ids[fmt.Sprintf("cj%d", i)] { + t.Fatalf("并发编号缺少 cj%d: %+v", i, ids) + } + } + + rollbackTx, err := db.Begin() + if err != nil { + t.Fatal(err) + } + rolledBackID, err := NextTaskID(rollbackTx, model.TaskPurchase) + if err != nil { + t.Fatal(err) + } + if err := rollbackTx.Rollback(); err != nil { + t.Fatal(err) + } + commitTx, err := db.Begin() + if err != nil { + t.Fatal(err) + } + committedID, err := NextTaskID(commitTx, model.TaskPurchase) + if err != nil { + t.Fatal(err) + } + if err := commitTx.Commit(); err != nil { + t.Fatal(err) + } + if rolledBackID != "cg1" || committedID != "cg1" { + t.Fatalf("回滚不应消耗采购序号: rollback=%s commit=%s", rolledBackID, committedID) + } +} + func openMySQLMigrationTestDB(t *testing.T) *sql.DB { t.Helper() if os.Getenv("CMAUTOBUY_MYSQL_TEST") != "1" { diff --git a/admin/repository/task.go b/admin/repository/task.go index 78a05a7..d3de472 100644 --- a/admin/repository/task.go +++ b/admin/repository/task.go @@ -5,11 +5,71 @@ import ( "database/sql" "errors" "fmt" + "strconv" "strings" "cmautobuy/admin/model" ) +const ( + collectTaskIDPrefix = "cj" + purchaseTaskIDPrefix = "cg" +) + +// NextTaskID 在当前事务内分配下一条任务业务单号。 +// +// 采集和采购分别递增；更新序列表后再读取同一行，依赖事务行锁保证并发不重复。 +// 调用方必须把分配和任务 INSERT 放在同一个事务里，失败回滚时不会消耗序号。 +func NextTaskID(q Execer, taskType model.TaskType) (string, error) { + prefix, ok := taskIDPrefix(taskType) + if !ok { + return "", fmt.Errorf("不支持为任务类型 %q 分配编号", taskType) + } + result, err := q.Exec(`UPDATE task_sequences SET current_value=current_value+1 WHERE task_type=?`, taskType) + if err != nil { + return "", fmt.Errorf("递增 %s 任务序号失败: %w", taskType, err) + } + affected, err := result.RowsAffected() + if err != nil { + return "", fmt.Errorf("读取 %s 任务序号更新结果失败: %w", taskType, err) + } + if affected != 1 { + return "", fmt.Errorf("%s 任务序列表缺失，请先完成数据库迁移", taskType) + } + var number int64 + if err := q.QueryRow(`SELECT current_value FROM task_sequences WHERE task_type=?`, taskType).Scan(&number); err != nil { + return "", fmt.Errorf("读取 %s 任务序号失败: %w", taskType, err) + } + if number <= 0 { + return "", fmt.Errorf("%s 任务序号不正确: %d", taskType, number) + } + return prefix + strconv.FormatInt(number, 10), nil +} + +func taskIDPrefix(taskType model.TaskType) (string, bool) { + switch taskType { + case model.TaskCollect: + return collectTaskIDPrefix, true + case model.TaskPurchase: + return purchaseTaskIDPrefix, true + default: + return "", false + } +} + +func parseTaskIDNumber(taskType model.TaskType, taskID string) (int64, bool) { + prefix, ok := taskIDPrefix(taskType) + if !ok || !strings.HasPrefix(taskID, prefix) || len(taskID) == len(prefix) { + return 0, false + } + numberText := taskID[len(prefix):] + if numberText[0] == '0' { + return 0, false + } + number, err := strconv.ParseInt(numberText, 10, 64) + return number, err == nil && number > 0 +} + // ClaimNextTask 为指定客户端领取一个任务。 // // 没有可领的任务时返回 (nil, nil) —— 调用方据此返回 204。 diff --git a/admin/service/pdd.go b/admin/service/pdd.go index 11ce6d0..5fc1b7c 100644 --- a/admin/service/pdd.go +++ b/admin/service/pdd.go @@ -647,8 +647,12 @@ func RecollectPddProductForUser(db *sql.DB, actor *model.User, goodsID, clientID return result, nil } + taskID, err := repository.NextTaskID(tx, model.TaskCollect) + if err != nil { + return result, err + } if err := repository.InsertCollectTaskForClientAndUser( - tx, newCollectTaskID(), p.GoodsID, p.URL, clientID, actor.UserID); err != nil { + tx, taskID, p.GoodsID, p.URL, clientID, actor.UserID); err != nil { return result, err } result.Created = 1 @@ -726,7 +730,10 @@ func createPddCollectTasks(db *sql.DB, goodsIDs []string, clientID, visibleUserI continue } - taskID := newCollectTaskID() + taskID, err := repository.NextTaskID(tx, model.TaskCollect) + if err != nil { + return CollectTaskResult{}, err + } if err := repository.InsertCollectTaskForClientAndUser( tx, taskID, p.GoodsID, p.URL, clientID, createdByUserID); err != nil { return CollectTaskResult{}, err @@ -808,16 +815,6 @@ func FormatCollectTaskMessage(r CollectTaskResult) string { return msg } -// newCollectTaskID 生成采集任务编号。 -// 带 COL- 前缀是为了在日志和数据库里一眼认出这是采集任务而不是采购任务。 -func newCollectTaskID() string { - id := newID() - if len(id) > 16 { - id = id[:16] - } - return "COL-" + id -} - // dedupe 去掉重复项并保持原有顺序，顺带丢掉空串。 func dedupe(values []string) []string { seen := make(map[string]bool, len(values)) diff --git a/admin/service/pdd_page_test.go b/admin/service/pdd_page_test.go index 598a3f5..e9c3412 100644 --- a/admin/service/pdd_page_test.go +++ b/admin/service/pdd_page_test.go @@ -610,6 +610,10 @@ func TestCreatePddCollectTasks_不指定客户端(t *testing.T) { if result.Created != 1 || result.Skipped() != 0 { t.Fatalf("Created=%d Skipped=%d，期望 1/0", result.Created, result.Skipped()) } + var taskID string + if err := db.QueryRow(`SELECT task_id FROM tasks WHERE task_type='collect' AND pdd_goods_id='737116531267'`).Scan(&taskID); err != nil || taskID != "cj1" { + t.Fatalf("第一条采集任务编号=%q err=%v，期望 cj1", taskID, err) + } status, assigned, url, ok := collectTaskOf(t, db, "737116531267") if !ok { diff --git a/admin/service/purchase_workflow.go b/admin/service/purchase_workflow.go index 45d91b4..db06d79 100644 --- a/admin/service/purchase_workflow.go +++ b/admin/service/purchase_workflow.go @@ -240,8 +240,12 @@ func CreatePurchaseTasksWithOptions(db *sql.DB, actor *model.User, requests []Pu result.Failures = append(result.Failures, TaskCreateError{SybID: request.SybID, Reason: reason}) continue } + taskID, err := repository.NextTaskID(tx, model.TaskPurchase) + if err != nil { + return PurchaseTaskResult{}, err + } if err := repository.InsertPurchaseTask(tx, model.Task{ - TaskID: newPurchaseTaskID(), AssignedClient: clientID, + TaskID: taskID, AssignedClient: clientID, CreatedByUserID: actor.UserID, ExecutionMode: executionMode, LiveConfirmedBy: confirmedBy, LiveConfirmedAt: confirmedAt, SybID: context.Order.SybID, OrderNo: context.Order.OrderNo, @@ -315,14 +319,6 @@ func validatePurchaseRequest(q repository.Execer, request PurchaseTaskRequest) ( return "", context, choice.OptionsJSON, nil } -func newPurchaseTaskID() string { - id := newID() - if len(id) > 16 { - id = id[:16] - } - return "PUR-" + id -} - // parsePriceYuanToCent 把网页输入的人民币元精确转成分，不使用浮点数。 func ParsePriceYuanToCent(raw string) (int64, error) { raw = strings.TrimSpace(raw) diff --git a/admin/service/purchase_workflow_test.go b/admin/service/purchase_workflow_test.go index 6e1f0d4..fd8c8cc 100644 --- a/admin/service/purchase_workflow_test.go +++ b/admin/service/purchase_workflow_test.go @@ -314,17 +314,17 @@ func TestCreatePurchaseTasks_安全字段和重复保护(t *testing.T) { t.Fatalf("创建结果=%+v err=%v", result, err) } var task model.Task - var status string - err = db.QueryRow(`SELECT status, assigned_client, syb_id, order_no, goods_id, + var status, taskID string + err = db.QueryRow(`SELECT task_id, status, assigned_client, syb_id, order_no, goods_id, shopee_sku_id, pdd_goods_url, pdd_goods_id, pdd_options, quantity, max_price_cent FROM tasks WHERE task_type='purchase' AND syb_id='SYB-1'`).Scan( - &status, &task.AssignedClient, &task.SybID, &task.OrderNo, &task.GoodsID, + &taskID, &status, &task.AssignedClient, &task.SybID, &task.OrderNo, &task.GoodsID, &task.ShopeeSKUID, &task.PddGoodsURL, &task.PddGoodsID, &task.PddOptions, &task.Quantity, &task.MaxPriceCent) if err != nil { t.Fatal(err) } - if status != "assigned" || task.MaxPriceCent != 4200 || task.MaxPriceCent == 999900 || + if taskID != "cg1" || status != "assigned" || task.MaxPriceCent != 4200 || task.MaxPriceCent == 999900 || task.Quantity != 1 || task.PddOptions != key || task.AssignedClient != "CLIENT-1" { t.Fatalf("采购任务字段不安全或不完整: status=%s task=%+v", status, task) } diff --git a/client/test/test_admin_gateway_contract.py b/client/test/test_admin_gateway_contract.py index f21bd38..0553338 100644 --- a/client/test/test_admin_gateway_contract.py +++ b/client/test/test_admin_gateway_contract.py @@ -108,10 +108,10 @@ class MockAdminGatewayContractTest(unittest.TestCase): def test_claim_respects_client_and_supported_types(self): self.gateway.enqueue_task( - self._task("COLLECT-001", TaskType.COLLECT), "client-001" + self._task("cj1", TaskType.COLLECT), "client-001" ) self.gateway.enqueue_task( - self._task("PURCHASE-001", TaskType.PURCHASE), "client-001" + self._task("cg1", TaskType.PURCHASE), "client-001" ) self.gateway.enqueue_task( self._task("COLLECT-OTHER", TaskType.COLLECT), "client-002" @@ -123,11 +123,11 @@ class MockAdminGatewayContractTest(unittest.TestCase): first = self.gateway.claim_next(self.client, collect_only) - self.assertEqual(first.task_id, "COLLECT-001") + self.assertEqual(first.task_id, "cj1") self.assertIsNone(self.gateway.claim_next(self.client, collect_only)) second = self.gateway.claim_next(self.client, self.all_capabilities) - self.assertEqual(second.task_id, "PURCHASE-001") + self.assertEqual(second.task_id, "cg1") self.assertIsNone( self.gateway.claim_next(self.client, self.all_capabilities) ) diff --git a/client/test/test_http_admin_gateway.py b/client/test/test_http_admin_gateway.py index 73b0704..d49d02a 100644 --- a/client/test/test_http_admin_gateway.py +++ b/client/test/test_http_admin_gateway.py @@ -225,7 +225,7 @@ class HttpAdminGatewayTest(unittest.TestCase): 200, { "task": { - "id": "COL-8020a8729f111c15", + "id": "cj1", "type": "collect", "version": 1, "priority": 0, @@ -250,7 +250,7 @@ class HttpAdminGatewayTest(unittest.TestCase): ) self.assertIsNotNone(task) - self.assertEqual(task.task_id, "COL-8020a8729f111c15") + self.assertEqual(task.task_id, "cj1") self.assertEqual(task.task_type.value, "collect") self.assertEqual(task.payload["goods_id"], "737116531267") self.assertEqual(opener.request.get_method(), "POST") diff --git a/docs/admin/01-requirements.md b/docs/admin/01-requirements.md index 0ced62b..609e8ab 100644 --- a/docs/admin/01-requirements.md +++ b/docs/admin/01-requirements.md @@ -253,6 +253,10 @@ PDD 商品之所以单独一个模块，是因为它在数据上就是**独立 操作员却只能盯着 PDD 商品的 `collect_status` 猜任务本身怎么样了 （被谁领了、领了多久、失败在哪一步），完全看不出任务这一层的信息。 +`task_id` 同时是页面业务单号和给 Client 的稳定主键：采集任务按 +`cj1`、`cj2` ……递增，采购任务按 `cg1`、`cg2` ……独立递增。任务类型 +仍以 `task_type` 为准，不允许仅根据编号前缀推导业务逻辑。 + `[必须]` 名字是「**采集采购**」，不是「任务」。它直接点出这一页装的是 哪两类任务，比泛称「任务」更能让操作员一眼知道点进去看什么。 diff --git a/docs/admin/03-data-model.md b/docs/admin/03-data-model.md index 5f85ce8..1fa7b1c 100644 --- a/docs/admin/03-data-model.md +++ b/docs/admin/03-data-model.md @@ -110,7 +110,9 @@ SQLite 同一时刻只允许一个写事务，连接放太开会互相抢锁、 生产 MySQL 迁移继续独立追加；v12（#165）新增 `task_syb_sources`，并把没有任何 有效采集任务的孤立 `pdd_products.collect_status='collecting'` 回收到 `pending`。 -历史 SQLite migrations 保持冻结，不追加该表。 +v13（#172）新增 `task_sequences`，把存量 `tasks.task_id` 按类型和创建时间确定性改为 +`cjN` / `cgN`，同步更新领取历史和顺运宝来源外键。历史 SQLite migrations +保持冻结，不追加 v12/v13 生产表结构。 **v3 为什么丢弃旧 `sku_mappings` 数据（见 #20）：** 新主键需要 `pdd_option_key`， 这是 Go 的 `service.OptionKey()` 用 `json.Marshal` 算出来的规范化键，SQL 语句 @@ -631,7 +633,7 @@ FROM spec_mapping_decisions WHERE suggested_option_key IS NULL GROUP BY rules_ve ```sql CREATE TABLE tasks ( - task_id TEXT PRIMARY KEY, -- 给 Client 的稳定编号，如 PDD-20260806-0001 + task_id TEXT PRIMARY KEY, -- 稳定业务主键：采集 cjN，采购 cgN task_type TEXT NOT NULL CHECK (task_type IN ('collect', 'purchase')), status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'assigned', 'claimed', @@ -671,6 +673,10 @@ CREATE INDEX idx_tasks_list ON tasks(updated_at DESC, task_id DESC); CREATE INDEX idx_tasks_order ON tasks(order_no); ``` +`task_id` 不是另外的展示别名，而是真实主键。采集和采购分别从 1 +开始递增，不共用计数器。分配编号与插入任务必须在同一事务中，以避免 +并发重号和创建失败消耗序号。Client 应把这个值当作不透明字符串原样保存。 + 生产 MySQL v5 在 `tasks` 追加以下安全字段（历史 SQLite 结构不改）： | 字段 | 含义 | @@ -703,7 +709,21 @@ CREATE INDEX idx_tasks_order ON tasks(order_no); 状态含义见 [01 需求](01-requirements.md) §6.2。 -### 7.1 `task_syb_sources` 采集任务来源（生产 MySQL v12） +### 7.1 `task_sequences` 任务序列（生产 MySQL v13） + +```sql +CREATE TABLE task_sequences ( + task_type VARCHAR(20) COLLATE utf8mb4_bin PRIMARY KEY, + current_value BIGINT NOT NULL DEFAULT 0, + CHECK (task_type IN ('collect', 'purchase')), + CHECK (current_value >= 0) +); +``` + +表中只有 `collect` 和 `purchase` 两行。创建任务时先在当前事务内对对应行 +`current_value + 1`，再读取并组成 `cjN` 或 `cgN`；InnoDB 行锁保证并发唯一。 + +### 7.2 `task_syb_sources` 采集任务来源（生产 MySQL v12） 从顺运宝创建采集任务时，同一个 PDD 商品可能对应多个货运单明细。来源使用关联表， 不能压进 `tasks.syb_id` 单列；后者仍只表示采购任务自身的顺运宝明细。 @@ -715,7 +735,7 @@ CREATE TABLE task_syb_sources ( created_at VARCHAR(35) NOT NULL, PRIMARY KEY (task_id, syb_id), KEY idx_task_syb_sources_syb (syb_id, task_id), - FOREIGN KEY (task_id) REFERENCES tasks(task_id) ON DELETE CASCADE, + FOREIGN KEY (task_id) REFERENCES tasks(task_id) ON DELETE CASCADE ON UPDATE CASCADE, FOREIGN KEY (syb_id) REFERENCES syb_orders(syb_id) ON DELETE CASCADE ); ``` @@ -735,12 +755,16 @@ CREATE TABLE task_claims ( task_id TEXT NOT NULL, client_id TEXT NOT NULL, claimed_at TEXT NOT NULL, - PRIMARY KEY (task_id, client_id) + PRIMARY KEY (task_id, client_id), + FOREIGN KEY (task_id) REFERENCES tasks(task_id) ON DELETE CASCADE ON UPDATE CASCADE ); CREATE INDEX idx_task_claims_client ON task_claims(client_id); ``` +生产 MySQL v13 清理了无对应任务的孤儿领取历史，并加入上述外键； +`ON UPDATE CASCADE` 保证主键迁移时领取历史同步改号。 + **为什么需要这张表：** [04 Client 接口实现](04-client-api.md) §4.1 要求 "只有**从未分配给该客户端**的任务才返回 403"。 但 `tasks.assigned_client` 只记**当前**归属，任务一旦重派给别人， diff --git a/docs/admin/04-client-api.md b/docs/admin/04-client-api.md index 0ff8931..f4f0061 100644 --- a/docs/admin/04-client-api.md +++ b/docs/admin/04-client-api.md @@ -17,6 +17,10 @@ POST /api/v1/client/tasks/{task_id}/result 提交成功结果 POST /api/v1/client/tasks/{task_id}/failure 提交失败/需人工 ``` +`task_id` 是不透明的稳定字符串，当前采集任务为 `cjN`、采购任务为 +`cgN`。Client 必须原样保存和回传，不校验旧前缀，不从编号推导类型； +业务类型始终以响应中的 `type` 为准。 + `[必须]` **不得新增"让 Client 查询状态"类接口**，也不得加回租约和心跳。 理由见 Client 契约 §1.1：本项目人工付款，重复下单只产生重复的**未付款**订单， 不值得为它引入一整套中断逻辑。 @@ -90,7 +94,7 @@ UPDATE tasks ```json { "task": { - "id": "PDD-20260806-0001", + "id": "cg1", "type": "purchase", "execution_mode": "dry_run", "version": 1, diff --git a/docs/admin/05-ui-specification.md b/docs/admin/05-ui-specification.md index 6457164..0f7e8fc 100644 --- a/docs/admin/05-ui-specification.md +++ b/docs/admin/05-ui-specification.md @@ -662,10 +662,14 @@ placeholder 写「任务编号 / 订单号 / 商品 ID」，**不要写全「PDD ```text ☐ │ 任务编号 │ 类型 │ 执行模式 │ 目标 │ PDD 店铺 │ 状态 │ 客户端 │ 更新时间 -☐ │ PDD-20260807-01 │ 采集 │ 采集 │ PDD 737116531267 │ 某某店铺 │ 已领取 │ 办公室-01 │ 15:20 -☐ │ PDD-20260807-02 │ 采购 │ 真实下单（不支付）│ SO-001 · 黑色/M · 2件 · ≤¥42.00 │ 某某店铺 │ 待领取 │ 办公室-02 │ 15:22 +☐ │ cj123 │ 采集 │ 采集 │ PDD 737116531267 │ 某某店铺 │ 已领取 │ 办公室-01 │ 15:20 +☐ │ cg45 │ 采购 │ 真实下单（不支付）│ SO-001 · 黑色/M · 2件 · ≤¥42.00 │ 某某店铺 │ 待领取 │ 办公室-02 │ 15:22 ``` +- `[必须]` **任务编号列**直接显示真实 `task_id`：采集为 `cjN`，采购为 + `cgN`；关键词支持完整或部分编号查询。“类型”列仍显示明确文字， + 不能只靠前缀或颜色区分。 + - `[必须]` **目标列**：采集显示 `PDD `（能 join 到未删除商品 的标题时追加显示）；采购显示 ` · <颜色/尺码> · <数量>件 · ≤<价格上限>`。 拼接逻辑在 service 层组装成一个字符串，模板只负责显示。 @@ -685,7 +689,7 @@ placeholder 写「任务编号 / 订单号 / 商品 ID」，**不要写全「PDD ```text ┌──────────────────────────────────────────────┐ -│ 任务 PDD-20260807-01 │ +│ 任务 cj123 │ ├──────────────────────────────────────────────┤ │ 类型 采集 │ │ 执行模式 采集 │ diff --git a/docs/client/04-admin-api-contract.md b/docs/client/04-admin-api-contract.md index 4056100..9e1925c 100644 --- a/docs/client/04-admin-api-contract.md +++ b/docs/client/04-admin-api-contract.md @@ -85,7 +85,7 @@ Idempotency-Key: ```json { - "id": "PDD-20260806-0001", + "id": "cg1", "type": "purchase", "version": 3, "priority": 10, @@ -165,6 +165,10 @@ X-Request-Id: ## 5. 领取任务 +`task.id` 是 Admin 分配的不透明稳定字符串，当前采集编号为 `cjN`、 +采购编号为 `cgN`。Client 必须原样写入 `remote_task_id` 并原样回传， +不校验历史前缀，不从编号推导类型或排序；任务类型以 `task.type` 为准。 + ```http POST /api/v1/client/tasks/claim ``` diff --git a/docs/client/05-ui-specification.md b/docs/client/05-ui-specification.md index 04bd398..5dfbd22 100644 --- a/docs/client/05-ui-specification.md +++ b/docs/client/05-ui-specification.md @@ -159,7 +159,7 @@ Client 首次显示主窗口时使用标准 Windows 最大化状态，不使用 - Admin 离线 —— 本地数据照常显示，只在信息条说明领取失败； - 加载失败。 -> **界面上说的"任务编号"一律指 `remote_task_id`**（例如 `PDD-20260806-0001`），不是数据库自增 `id`，更不是表格行号。行号会随排序和筛选变化，拿它当任务身份必出错。 +> **界面上说的"任务编号"一律指 `remote_task_id`**（例如采集 `cj1`、采购 `cg1`），不是数据库自增 `id`，更不是表格行号。行号会随排序和筛选变化，拿它当任务身份必出错。 ### 5.3 增量加载模板（照抄即可）