From 13cfced80db1a233ae8228687588243a767c37e8 Mon Sep 17 00:00:00 2001 From: chengma Date: Mon, 10 Aug 2026 02:21:31 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=BF=81=E7=A7=BB=20Admin=20Repository?= =?UTF-8?q?=20=E5=88=B0=20MySQL=208=20(#79)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- admin/auth_integration_test.go | 20 +---- admin/client_api_auth_regression_test.go | 11 +-- admin/client_assignment_integration_test.go | 9 +-- admin/internal/testutil/mysql.go | 88 ++++++++++++++++++++ admin/main.go | 17 ++-- admin/main_test.go | 9 +-- admin/mysql_test_helpers_test.go | 13 +++ admin/password_change_integration_test.go | 9 +-- admin/repository/client.go | 37 +++++---- admin/repository/idempotency.go | 16 +++- admin/repository/mapping.go | 12 +-- admin/repository/migrate_test.go | 2 +- admin/repository/mysql_db.go | 39 ++++++++- admin/repository/pdd.go | 21 +++-- admin/repository/shopee.go | 30 +++---- admin/repository/syb.go | 43 +++++----- admin/repository/syb_test.go | 51 ++++++++++-- admin/repository/task.go | 89 +++++++++------------ admin/repository/user.go | 35 +++++--- admin/service/client_test.go | 12 +-- admin/service/pdd_page_test.go | 4 +- admin/service/pdd_test.go | 2 +- admin/service/purchase_workflow_test.go | 10 ++- admin/service/submit.go | 16 ++++ admin/service/syb_test.go | 10 +-- admin/user_integration_test.go | 9 +-- docs/admin/00-glossary.md | 4 +- docs/admin/05-ui-specification.md | 6 +- docs/admin/08-顺运宝接口.md | 6 +- 29 files changed, 390 insertions(+), 240 deletions(-) create mode 100644 admin/internal/testutil/mysql.go create mode 100644 admin/mysql_test_helpers_test.go diff --git a/admin/auth_integration_test.go b/admin/auth_integration_test.go index a67db64..4e1a1aa 100644 --- a/admin/auth_integration_test.go +++ b/admin/auth_integration_test.go @@ -6,19 +6,10 @@ import ( "net/url" "strings" "testing" - - "cmautobuy/admin/repository" ) func TestAdmin首次初始化登录退出完整流程(t *testing.T) { - db, err := repository.Open(t.TempDir()) - if err != nil { - t.Fatal(err) - } - defer db.Close() - if err := repository.Migrate(db); err != nil { - t.Fatal(err) - } + db := newMySQLTestDB(t) router, err := newRouter(db) if err != nil { t.Fatal(err) @@ -118,14 +109,7 @@ func TestAdmin首次初始化登录退出完整流程(t *testing.T) { } func TestClientAPI不经过网页登录和CSRF(t *testing.T) { - db, err := repository.Open(t.TempDir()) - if err != nil { - t.Fatal(err) - } - defer db.Close() - if err := repository.Migrate(db); err != nil { - t.Fatal(err) - } + db := newMySQLTestDB(t) router, err := newRouter(db) if err != nil { t.Fatal(err) diff --git a/admin/client_api_auth_regression_test.go b/admin/client_api_auth_regression_test.go index 3db72cf..685d709 100644 --- a/admin/client_api_auth_regression_test.go +++ b/admin/client_api_auth_regression_test.go @@ -11,7 +11,6 @@ import ( "time" "cmautobuy/admin/model" - "cmautobuy/admin/repository" "cmautobuy/admin/service" ) @@ -166,15 +165,7 @@ func TestClient提交HTTP契约_取消重派与幂等不受网页登录影响(t func newAPIRouteTestDB(t *testing.T) *sql.DB { t.Helper() - db, err := repository.Open(t.TempDir()) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { db.Close() }) - if err := repository.Migrate(db); err != nil { - t.Fatal(err) - } - return db + return newMySQLTestDB(t) } func insertAPIRouteTask(t *testing.T, db *sql.DB, taskID, clientID string, createdAt time.Time) { diff --git a/admin/client_assignment_integration_test.go b/admin/client_assignment_integration_test.go index 51c2be9..2bfebd8 100644 --- a/admin/client_assignment_integration_test.go +++ b/admin/client_assignment_integration_test.go @@ -14,14 +14,7 @@ import ( ) func Test客户端归属页面与管理员写权限(t *testing.T) { - db, err := repository.Open(t.TempDir()) - if err != nil { - t.Fatal(err) - } - defer db.Close() - if err := repository.Migrate(db); err != nil { - t.Fatal(err) - } + db := newMySQLTestDB(t) now := time.Now() if err := service.SetupInitialAdmin(db, "admin", "admin-password", "admin-password", now); err != nil { t.Fatal(err) diff --git a/admin/internal/testutil/mysql.go b/admin/internal/testutil/mysql.go new file mode 100644 index 0000000..ddc01c1 --- /dev/null +++ b/admin/internal/testutil/mysql.go @@ -0,0 +1,88 @@ +// Package testutil 为 Admin 测试创建彼此隔离的 MySQL 8 数据库。 +package testutil + +import ( + "crypto/rand" + "database/sql" + "encoding/hex" + "fmt" + "os" + "strings" + + "cmautobuy/admin/config" + "cmautobuy/admin/repository" +) + +// TestingT 是本包需要的 testing.T 最小接口，避免生产包依赖 testing。 +type TestingT interface { + Helper() + Cleanup(func()) + Fatalf(format string, args ...any) + Errorf(format string, args ...any) + Skipf(format string, args ...any) +} + +// OpenMySQL 为一个测试创建独立数据库，测试结束后只删除该数据库。 +func OpenMySQL(t TestingT) *sql.DB { + t.Helper() + if os.Getenv("CMAUTOBUY_MYSQL_TEST") != "1" { + t.Skipf("未设置 CMAUTOBUY_MYSQL_TEST=1，跳过真实 MySQL 8 测试") + } + baseConfig, err := config.LoadDatabaseFromEnv() + if err != nil { + t.Fatalf("读取 MySQL 测试配置失败: %v", err) + } + if !strings.HasSuffix(baseConfig.Name, "_test") { + t.Fatalf("MySQL 测试基础库名必须以 _test 结尾，实际 %q", baseConfig.Name) + } + adminDB, err := repository.OpenMySQL(baseConfig) + if err != nil { + t.Fatalf("连接 MySQL 测试基础库失败: %v", err) + } + + databaseName := testDatabaseName(baseConfig.Name) + quotedName := quoteIdentifier(databaseName) + if _, err := adminDB.Exec("CREATE DATABASE " + quotedName + + " CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci"); err != nil { + adminDB.Close() + t.Fatalf("创建隔离 MySQL 测试库失败: %v", err) + } + + testConfig := baseConfig + testConfig.Name = databaseName + db, err := repository.OpenMySQL(testConfig) + if err != nil { + adminDB.Exec("DROP DATABASE " + quotedName) + adminDB.Close() + t.Fatalf("连接隔离 MySQL 测试库失败: %v", err) + } + if err := repository.MigrateMySQL(db); err != nil { + db.Close() + adminDB.Exec("DROP DATABASE " + quotedName) + adminDB.Close() + t.Fatalf("初始化隔离 MySQL 测试库失败: %v", err) + } + t.Cleanup(func() { + db.Close() + if _, err := adminDB.Exec("DROP DATABASE " + quotedName); err != nil { + t.Errorf("清理隔离 MySQL 测试库 %s 失败: %v", databaseName, err) + } + adminDB.Close() + }) + return db +} + +func testDatabaseName(base string) string { + random := make([]byte, 8) + if _, err := rand.Read(random); err != nil { + panic(fmt.Sprintf("生成 MySQL 测试库随机名失败: %v", err)) + } + if len(base) > 40 { + base = base[:40] + } + return base + "_" + hex.EncodeToString(random) +} + +func quoteIdentifier(name string) string { + return "`" + strings.ReplaceAll(name, "`", "``") + "`" +} diff --git a/admin/main.go b/admin/main.go index 4677bfc..2ef0dce 100644 --- a/admin/main.go +++ b/admin/main.go @@ -48,18 +48,19 @@ func main() { } log.Printf("数据目录: %s", dataDir) - // 2. 数据库 - db, err := repository.Open(dataDir) + // 2. 数据库。凭据只从环境变量读取，不写进配置文件或日志。 + databaseConfig, err := config.LoadDatabaseFromEnv() if err != nil { - log.Fatalf("打开数据库失败: %v", err) + log.Fatalf("读取 MySQL 配置失败: %v", err) + } + db, err := repository.OpenMySQL(databaseConfig) + if err != nil { + log.Fatalf("打开 MySQL 数据库失败: %v", err) } defer db.Close() - if err := repository.Migrate(db); err != nil { - log.Fatalf("数据库迁移失败: %v", err) - } - if err := repository.CheckSchema(db); err != nil { - log.Fatalf("%v", err) + if err := repository.MigrateMySQL(db); err != nil { + log.Fatalf("MySQL 数据库迁移失败: %v", err) } interrupted, err := service.InterruptRunningSybSyncRuns(db, time.Now()) if err != nil { diff --git a/admin/main_test.go b/admin/main_test.go index e8ebf19..fd5aef2 100644 --- a/admin/main_test.go +++ b/admin/main_test.go @@ -38,14 +38,7 @@ func TestPaginationCSS_桌面右对齐且窄窗口不遮挡内容(t *testing.T) // 六个主页面都走一次真实路由和模板渲染。 // 这样模板字段写错或新增列漏接时，测试阶段就会失败，不必等人工点页面。 func TestMainPagesReturnOK(t *testing.T) { - db, err := repository.Open(t.TempDir()) - if err != nil { - t.Fatalf("打开测试数据库失败: %v", err) - } - defer db.Close() - if err := repository.Migrate(db); err != nil { - t.Fatalf("迁移测试数据库失败: %v", err) - } + db := newMySQLTestDB(t) router, err := newRouter(db) if err != nil { diff --git a/admin/mysql_test_helpers_test.go b/admin/mysql_test_helpers_test.go new file mode 100644 index 0000000..7b10698 --- /dev/null +++ b/admin/mysql_test_helpers_test.go @@ -0,0 +1,13 @@ +package main + +import ( + "database/sql" + "testing" + + "cmautobuy/admin/internal/testutil" +) + +func newMySQLTestDB(t *testing.T) *sql.DB { + t.Helper() + return testutil.OpenMySQL(t) +} diff --git a/admin/password_change_integration_test.go b/admin/password_change_integration_test.go index 791d53d..2f95725 100644 --- a/admin/password_change_integration_test.go +++ b/admin/password_change_integration_test.go @@ -14,14 +14,7 @@ import ( ) func Test管理员自助修改密码完整流程和权限边界(t *testing.T) { - db, err := repository.Open(t.TempDir()) - if err != nil { - t.Fatal(err) - } - defer db.Close() - if err := repository.Migrate(db); err != nil { - t.Fatal(err) - } + db := newMySQLTestDB(t) now := time.Now() if err := service.SetupInitialAdmin(db, "admin", "old-password", "old-password", now); err != nil { t.Fatal(err) diff --git a/admin/repository/client.go b/admin/repository/client.go index c5909fc..bafa683 100644 --- a/admin/repository/client.go +++ b/admin/repository/client.go @@ -55,14 +55,14 @@ func UpsertClient(q Execer, c model.Client, explicit bool) error { pdd_package, capabilities, last_seen_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(client_id) DO UPDATE SET + ON DUPLICATE KEY UPDATE name = CASE WHEN ? THEN ? ELSE clients.name END, - device_address = excluded.device_address, - platform = excluded.platform, - pdd_package = excluded.pdd_package, - capabilities = excluded.capabilities, - last_seen_at = excluded.last_seen_at, - updated_at = excluded.updated_at`, + device_address = VALUES(device_address), + platform = VALUES(platform), + pdd_package = VALUES(pdd_package), + capabilities = VALUES(capabilities), + last_seen_at = VALUES(last_seen_at), + updated_at = VALUES(updated_at)`, c.ClientID, insertName, c.DeviceAddress, c.Platform, c.PddPackage, c.Capabilities, now, now, now, updateName, name) @@ -155,8 +155,11 @@ func listClientsForUserPage(db *sql.DB, keyword, visibleUserID string, limit, of query := `SELECT c.client_id, c.name, c.device_address, c.platform, c.pdd_package, c.capabilities, c.last_seen_at, c.created_at, c.updated_at, a.user_id, u.username` + clientListFrom + where + - ` ORDER BY c.last_seen_at DESC, c.client_id LIMIT ? OFFSET ?` - args = append(args, limit, offset) + ` ORDER BY c.last_seen_at DESC, c.client_id` + if limit >= 0 { + query += ` LIMIT ? OFFSET ?` + args = append(args, limit, offset) + } rows, err := db.Query(query, args...) if err != nil { @@ -216,16 +219,16 @@ func ListClientsForUserPage(db *sql.DB, keyword, visibleUserID string, limit, of } // CountOnlineClientsForUser 统计同一筛选和权限范围内的在线客户端。 -// julianday 解析失败会得到 NULL，自然按离线处理，与 model.Client.IsOnline 一致。 +// last_seen_at 统一以 UTC RFC3339 保存，固定宽度字符串的字典序就是时间顺序。 func CountOnlineClientsForUser(db *sql.DB, keyword, visibleUserID, cutoffISO string) (int, error) { where, args := clientListFilter(keyword, visibleUserID) args = append(args, cutoffISO) var online int query := `SELECT COUNT(*)` + clientListFrom + where if where == "" { - query += ` WHERE julianday(c.last_seen_at) > julianday(?)` + query += ` WHERE c.last_seen_at > ?` } else { - query += ` AND julianday(c.last_seen_at) > julianday(?)` + query += ` AND c.last_seen_at > ?` } if err := db.QueryRow(query, args...).Scan(&online); err != nil { return 0, fmt.Errorf("统计在线客户端失败: %w", err) @@ -262,13 +265,13 @@ func AssignClient(db *sql.DB, assignment model.ClientUserAssignment) (bool, bool return false, false, fmt.Errorf("开始绑定客户端事务失败: %w", err) } defer tx.Rollback() - var exists int - if err := tx.QueryRow(`SELECT COUNT(*) FROM clients WHERE client_id = ?`, assignment.ClientID).Scan(&exists); err != nil { + var lockedClientID string + if err := tx.QueryRow(`SELECT client_id FROM clients WHERE client_id = ? FOR UPDATE`, assignment.ClientID).Scan(&lockedClientID); errors.Is(err, sql.ErrNoRows) { + return false, false, ErrClientNotFound + } else if err != nil { return false, false, fmt.Errorf("检查客户端失败: %w", err) } - if exists == 0 { - return false, false, ErrClientNotFound - } + var exists int if err := tx.QueryRow(`SELECT COUNT(*) FROM users WHERE user_id = ? AND role = ? AND status = ?`, assignment.UserID, model.RolePurchaser, model.UserActive).Scan(&exists); err != nil { return false, false, fmt.Errorf("检查采购员失败: %w", err) diff --git a/admin/repository/idempotency.go b/admin/repository/idempotency.go index 80e276a..4d52c35 100644 --- a/admin/repository/idempotency.go +++ b/admin/repository/idempotency.go @@ -8,13 +8,17 @@ import ( "fmt" "cmautobuy/admin/model" + "github.com/go-sql-driver/mysql" ) // ErrIdempotencyConflict 表示同一个键提交了不同的内容。 // // 这说明客户端弄错了——同一个键必须对应同一份内容。 // 内容真的变了，应该用新的 attempt_id 生成新的键。 -var ErrIdempotencyConflict = errors.New("相同幂等键提交了不同内容") +var ( + ErrIdempotencyConflict = errors.New("相同幂等键提交了不同内容") + ErrIdempotencyAlreadySaved = errors.New("幂等键已被并发请求保存") +) // HashRequest 计算请求体的哈希，用来判断"同一个键"配的是不是"同一份内容"。 func HashRequest(body []byte) string { @@ -31,7 +35,7 @@ func HashRequest(body []byte) string { func LookupIdempotent(q Execer, key, requestHash string) (string, bool, error) { var storedHash, storedBody string err := q.QueryRow( - `SELECT request_hash, response_body FROM idempotency_keys WHERE key = ?`, + "SELECT request_hash, response_body FROM idempotency_keys WHERE `key` = ?", key).Scan(&storedHash, &storedBody) if err == sql.ErrNoRows { @@ -52,10 +56,14 @@ func LookupIdempotent(q Execer, key, requestHash string) (string, bool, error) { // 分开写的话，业务写成功但幂等记录没写上，客户端重试就会被重复处理。 func SaveIdempotent(q Execer, key, requestHash, responseBody string) error { _, err := q.Exec( - `INSERT INTO idempotency_keys (key, request_hash, response_body, created_at) - VALUES (?, ?, ?, ?)`, + "INSERT INTO idempotency_keys (`key`, request_hash, response_body, created_at)\n"+ + " VALUES (?, ?, ?, ?)", key, requestHash, responseBody, model.NowISO()) if err != nil { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { + return ErrIdempotencyAlreadySaved + } return fmt.Errorf("保存幂等键失败: %w", err) } return nil diff --git a/admin/repository/mapping.go b/admin/repository/mapping.go index 8269f4f..d2b36d5 100644 --- a/admin/repository/mapping.go +++ b/admin/repository/mapping.go @@ -37,12 +37,12 @@ func UpsertSKUMapping(q Execer, m model.SKUMapping) error { (shopee_sku_id, pdd_goods_id, pdd_option_key, pdd_options, goods_id, mapped_at, mapped_by) VALUES (?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(shopee_sku_id, pdd_goods_id) DO UPDATE SET - pdd_option_key = excluded.pdd_option_key, - pdd_options = excluded.pdd_options, - goods_id = excluded.goods_id, - mapped_at = excluded.mapped_at, - mapped_by = excluded.mapped_by`, + ON DUPLICATE KEY UPDATE + pdd_option_key = VALUES(pdd_option_key), + pdd_options = VALUES(pdd_options), + goods_id = VALUES(goods_id), + mapped_at = VALUES(mapped_at), + mapped_by = VALUES(mapped_by)`, m.ShopeeSKUID, m.PddGoodsID, m.PddOptionKey, m.PddOptions, m.GoodsID, m.MappedAt, nullableText(m.MappedBy)) if err != nil { diff --git a/admin/repository/migrate_test.go b/admin/repository/migrate_test.go index 662dd12..0bb2722 100644 --- a/admin/repository/migrate_test.go +++ b/admin/repository/migrate_test.go @@ -665,7 +665,7 @@ func TestMigrate_v8新增顺运宝同步记录表(t *testing.T) { Role: model.RolePurchaser, Status: model.UserActive, PasswordChangedAt: model.NowISO(), CreatedAt: model.NowISO(), UpdatedAt: model.NowISO(), } - if err := CreateInitialAdmin(db, user); err != nil { + if err := CreateUser(db, user); err != nil { t.Fatalf("准备同步记录外键用户失败: %v", err) } // 状态和 running/finished_at 对应关系必须由数据库兜底，不能只靠 Go 校验。 diff --git a/admin/repository/mysql_db.go b/admin/repository/mysql_db.go index d549fcd..a5436b4 100644 --- a/admin/repository/mysql_db.go +++ b/admin/repository/mysql_db.go @@ -13,7 +13,7 @@ import ( "cmautobuy/admin/config" ) -const mysqlSchemaVersion = 1 +const mysqlSchemaVersion = 2 // OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。 func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) { @@ -29,6 +29,8 @@ func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) { driverConfig.ReadTimeout = 30 * time.Second driverConfig.WriteTimeout = 30 * time.Second driverConfig.RejectReadOnly = true + // 业务层用 RowsAffected 判断目标行是否存在；重复写入相同值也应算匹配到。 + driverConfig.ClientFoundRows = true driverConfig.Params = map[string]string{ "time_zone": "'+00:00'", "sql_mode": "'STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION'", @@ -298,6 +300,17 @@ var mysqlSchemaV1 = []string{ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, } +// mysqlSchemaV2 增加首位管理员初始化的并发哨兵。MySQL DDL 会隐式提交， +// 因此每条语句都必须可重放，并在全部成功后才记录版本。 +var mysqlSchemaV2 = []string{ + `CREATE TABLE IF NOT EXISTS admin_initialization_lock ( + id TINYINT PRIMARY KEY, + CHECK (id = 1) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, + `INSERT INTO admin_initialization_lock (id) VALUES (1) + ON DUPLICATE KEY UPDATE id = VALUES(id)`, +} + // MigrateMySQL 建立或升级 MySQL schema。生产迁移只能在这里追加新版本。 func MigrateMySQL(db *sql.DB) error { if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations ( @@ -319,19 +332,39 @@ func MigrateMySQL(db *sql.DB) error { return fmt.Errorf("执行 MySQL schema v1 第 %d 条失败: %w", i+1, err) } } - if err := CheckMySQLSchema(db); err != nil { + if err := checkMySQLSchema(db, requiredTables); err != nil { return fmt.Errorf("MySQL schema v1 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 1, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v1 失败: %w", err) } + current = 1 + } + if current < 2 { + for i, statement := range mysqlSchemaV2 { + if _, err := db.Exec(statement); err != nil { + return fmt.Errorf("执行 MySQL schema v2 第 %d 条失败: %w", i+1, err) + } + } + if err := CheckMySQLSchema(db); err != nil { + return fmt.Errorf("MySQL schema v2 自检失败，未记录版本: %w", err) + } + if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, + 2, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { + return fmt.Errorf("记录 MySQL schema v2 失败: %w", err) + } } return CheckMySQLSchema(db) } // CheckMySQLSchema 确认所有业务表和关键追加列存在。 func CheckMySQLSchema(db *sql.DB) error { + mysqlRequiredTables := append(append([]string{}, requiredTables...), "admin_initialization_lock") + return checkMySQLSchema(db, mysqlRequiredTables) +} + +func checkMySQLSchema(db *sql.DB, tables []string) error { rows, err := db.Query(`SELECT table_name FROM information_schema.tables WHERE table_schema = DATABASE() AND table_type = 'BASE TABLE'`) if err != nil { @@ -350,7 +383,7 @@ func CheckMySQLSchema(db *sql.DB) error { return err } var missing []string - for _, table := range requiredTables { + for _, table := range tables { if !existing[table] { missing = append(missing, table) } diff --git a/admin/repository/pdd.go b/admin/repository/pdd.go index 761b5c9..25b39ad 100644 --- a/admin/repository/pdd.go +++ b/admin/repository/pdd.go @@ -163,15 +163,15 @@ type PddProductRow struct { SkuCount int } -// 规格数交给 SQLite 算，不要把整个 skus_json 读进 Go 再数—— +// 规格数交给 MySQL 算，不要把整个 skus_json 读进 Go 再数—— // 列表页几十上百行，每行都反序列化一遍纯属浪费。 // -// json_valid 那层判断不能省：skus_json 万一存进了坏数据， -// json_array_length 会让**整条查询报错**，页面直接打不开。 +// JSON_VALID 那层判断不能省：skus_json 万一存进了坏数据， +// JSON_LENGTH 会让**整条查询报错**，页面直接打不开。 const pddSkuCountExpr = ` - CASE WHEN skus_json IS NULL OR skus_json = '' OR NOT json_valid(skus_json) + CASE WHEN skus_json IS NULL OR skus_json = '' OR NOT JSON_VALID(skus_json) THEN -1 - ELSE COALESCE(json_array_length(skus_json, '$.skus'), -1) + ELSE COALESCE(JSON_LENGTH(skus_json, '$.skus'), -1) END` // pddListFilter 统一生成 PDD 列表和总数查询的筛选条件。 @@ -181,7 +181,7 @@ func pddListFilter(keyword string, status model.CollectStatus) (string, []any) { args := make([]any, 0, 3) if keyword = strings.TrimSpace(keyword); keyword != "" { pattern := "%" + escapeLike(keyword) + "%" - where += ` AND (goods_id LIKE ? ESCAPE '\' OR url LIKE ? ESCAPE '\')` + where += ` AND (goods_id LIKE ? ESCAPE '!' OR url LIKE ? ESCAPE '!')` args = append(args, pattern, pattern) } if status != "" { @@ -230,12 +230,11 @@ func ListPddProducts(q Execer, keyword string, status model.CollectStatus, limit return list, total, nil } -// escapeLike 把 LIKE 的三个特殊字符转义掉，配合 ESCAPE '\' 使用。 -// 反斜杠必须第一个换，否则会把后面刚加的反斜杠又转义一遍。 +// escapeLike 把 LIKE 的三个特殊字符转义掉，配合 ESCAPE '!' 使用。 func escapeLike(s string) string { - s = strings.ReplaceAll(s, `\`, `\\`) - s = strings.ReplaceAll(s, `%`, `\%`) - s = strings.ReplaceAll(s, `_`, `\_`) + s = strings.ReplaceAll(s, `!`, `!!`) + s = strings.ReplaceAll(s, `%`, `!%`) + s = strings.ReplaceAll(s, `_`, `!_`) return s } diff --git a/admin/repository/shopee.go b/admin/repository/shopee.go index 7f8ad91..c005ec6 100644 --- a/admin/repository/shopee.go +++ b/admin/repository/shopee.go @@ -24,11 +24,11 @@ func UpsertShopeeProduct(q Execer, goodsID, title, shopeeStatus, mainSKUCode str _, err := q.Exec(` INSERT INTO shopee_products (goods_id, title, shopee_status, main_sku_code, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?) - ON CONFLICT(goods_id) DO UPDATE SET - title = excluded.title, - shopee_status = excluded.shopee_status, - main_sku_code = excluded.main_sku_code, - updated_at = excluded.updated_at`, + ON DUPLICATE KEY UPDATE + title = VALUES(title), + shopee_status = VALUES(shopee_status), + main_sku_code = VALUES(main_sku_code), + updated_at = VALUES(updated_at)`, // 注意：pdd_goods_url / pdd_goods_id 两列不在 INSERT 的列清单里， // 也不在 DO UPDATE SET 里——新建时它们是 NULL（未填链接）， // 已存在时它们完全不受这条语句影响。 @@ -61,15 +61,15 @@ func UpsertShopeeSKU(q Execer, skuID, goodsID, specRaw, color, size, advice stri (sku_id, goods_id, spec_raw, color, size, advice, parse_ok, sku_code, is_manual, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?) - ON CONFLICT(sku_id) DO UPDATE SET - goods_id = excluded.goods_id, - spec_raw = excluded.spec_raw, - color = excluded.color, - size = excluded.size, - advice = excluded.advice, - parse_ok = excluded.parse_ok, - sku_code = excluded.sku_code, - updated_at = excluded.updated_at`, + ON DUPLICATE KEY UPDATE + goods_id = VALUES(goods_id), + spec_raw = VALUES(spec_raw), + color = VALUES(color), + size = VALUES(size), + advice = VALUES(advice), + parse_ok = VALUES(parse_ok), + sku_code = VALUES(sku_code), + updated_at = VALUES(updated_at)`, // is_manual 不在上面的 SET 列表里，SQLite 对没提到的列保持原值不变。 skuID, goodsID, specRaw, color, size, advice, parseOKInt, skuCode, now, now) if err != nil { @@ -132,7 +132,7 @@ func shopeeFilterClause(filter ShopeeFilter) (string, []any) { if kw := strings.TrimSpace(filter.Keyword); kw != "" { like := "%" + escapeLike(kw) + "%" - clauses = append(clauses, `(sp.goods_id LIKE ? ESCAPE '\' OR sp.title LIKE ? ESCAPE '\')`) + clauses = append(clauses, `(sp.goods_id LIKE ? ESCAPE '!' OR sp.title LIKE ? ESCAPE '!')`) args = append(args, like, like) } diff --git a/admin/repository/syb.go b/admin/repository/syb.go index 4a35633..18957ba 100644 --- a/admin/repository/syb.go +++ b/admin/repository/syb.go @@ -29,10 +29,10 @@ func SaveSybSession(q Execer, username, cookiesJSON, expiresAt string) error { _, err := q.Exec(` INSERT INTO syb_session (username, cookies, expires_at, updated_at) VALUES (?, ?, ?, ?) - ON CONFLICT(username) DO UPDATE SET - cookies = excluded.cookies, - expires_at = excluded.expires_at, - updated_at = excluded.updated_at`, + ON DUPLICATE KEY UPDATE + cookies = VALUES(cookies), + expires_at = VALUES(expires_at), + updated_at = VALUES(updated_at)`, username, cookiesJSON, expiresAt, now) if err != nil { return fmt.Errorf("保存顺运宝会话缓存失败: %w", err) @@ -103,9 +103,9 @@ func SetSybLastSyncedAt(q Execer, at string) error { _, err := q.Exec(` INSERT INTO syb_sync_state (id, last_synced_at, updated_at) VALUES (1, ?, ?) - ON CONFLICT(id) DO UPDATE SET - last_synced_at = excluded.last_synced_at, - updated_at = excluded.updated_at`, + ON DUPLICATE KEY UPDATE + last_synced_at = VALUES(last_synced_at), + updated_at = VALUES(updated_at)`, at, now) if err != nil { return fmt.Errorf("更新顺运宝同步进度失败: %w", err) @@ -254,16 +254,16 @@ func UpsertSybOrder(q Execer, o model.SybOrder) (created bool, err error) { (syb_id, order_no, title, product_spec, shopee_goods_id, shopee_sku_id, quantity, price_twd_cent, image_url, syb_data, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(syb_id) DO UPDATE SET - order_no = excluded.order_no, - title = excluded.title, - product_spec = excluded.product_spec, - shopee_goods_id = excluded.shopee_goods_id, - quantity = excluded.quantity, - price_twd_cent = excluded.price_twd_cent, - image_url = excluded.image_url, - syb_data = excluded.syb_data, - updated_at = excluded.updated_at`, + ON DUPLICATE KEY UPDATE + order_no = VALUES(order_no), + title = VALUES(title), + product_spec = VALUES(product_spec), + shopee_goods_id = VALUES(shopee_goods_id), + quantity = VALUES(quantity), + price_twd_cent = VALUES(price_twd_cent), + image_url = VALUES(image_url), + syb_data = VALUES(syb_data), + updated_at = VALUES(updated_at)`, // 注意：shopee_sku_id 只出现在 INSERT 的列清单里（新建行时写 o.ShopeeSKUID， // 同步永远传空字符串），完全不出现在 DO UPDATE SET 里——已存在的行 // 这一列不受本语句影响，见上面的函数注释。 @@ -297,7 +297,7 @@ func sybOrderFilterClause(filter SybOrderFilter) (string, []any) { var args []any if kw := filter.Keyword; kw != "" { like := "%" + escapeLike(kw) + "%" - clauses = append(clauses, `(so.order_no LIKE ? ESCAPE '\' OR so.title LIKE ? ESCAPE '\')`) + clauses = append(clauses, `(so.order_no LIKE ? ESCAPE '!' OR so.title LIKE ? ESCAPE '!')`) args = append(args, like, like) } switch filter.Stage { @@ -386,8 +386,11 @@ func ListSybOrderContexts(q Execer, filter SybOrderFilter, limit, offset int) ([ AND t.syb_id = so.syb_id AND t.status IN ('pending', 'assigned', 'claimed'))` + sybOrderContextFrom + where + ` - ORDER BY so.updated_at DESC, so.syb_id DESC LIMIT ? OFFSET ?` - args = append(args, limit, offset) + ORDER BY so.updated_at DESC, so.syb_id DESC` + if limit >= 0 { + query += ` LIMIT ? OFFSET ?` + args = append(args, limit, offset) + } rows, err := q.Query(query, args...) if err != nil { return nil, fmt.Errorf("查询顺运宝采购处理列表失败: %w", err) diff --git a/admin/repository/syb_test.go b/admin/repository/syb_test.go index 149a352..87a5ebc 100644 --- a/admin/repository/syb_test.go +++ b/admin/repository/syb_test.go @@ -1,22 +1,63 @@ package repository import ( + "crypto/rand" "database/sql" + "encoding/hex" + "os" + "strings" "testing" + "cmautobuy/admin/config" "cmautobuy/admin/model" ) func newSybTestDB(t *testing.T) *sql.DB { t.Helper() - db, err := Open(t.TempDir()) + if os.Getenv("CMAUTOBUY_MYSQL_TEST") != "1" { + t.Skip("未启用真实 MySQL 8 集成测试") + } + cfg, err := config.LoadDatabaseFromEnv() if err != nil { - t.Fatalf("打开测试库失败: %v", err) + t.Fatal(err) } - t.Cleanup(func() { db.Close() }) - if err := Migrate(db); err != nil { - t.Fatalf("迁移失败: %v", err) + if !strings.HasSuffix(cfg.Name, "_test") { + t.Fatalf("MySQL 测试基础库名必须以 _test 结尾") } + adminDB, err := OpenMySQL(cfg) + if err != nil { + t.Fatal(err) + } + random := make([]byte, 8) + if _, err := rand.Read(random); err != nil { + t.Fatal(err) + } + databaseName := cfg.Name + "_" + hex.EncodeToString(random) + quotedName := "`" + strings.ReplaceAll(databaseName, "`", "``") + "`" + if _, err := adminDB.Exec("CREATE DATABASE " + quotedName + " CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci"); err != nil { + adminDB.Close() + t.Fatal(err) + } + cfg.Name = databaseName + db, err := OpenMySQL(cfg) + if err != nil { + adminDB.Exec("DROP DATABASE " + quotedName) + adminDB.Close() + t.Fatal(err) + } + if err := MigrateMySQL(db); err != nil { + db.Close() + adminDB.Exec("DROP DATABASE " + quotedName) + adminDB.Close() + t.Fatal(err) + } + t.Cleanup(func() { + db.Close() + if _, err := adminDB.Exec("DROP DATABASE " + quotedName); err != nil { + t.Errorf("清理隔离 MySQL 测试库失败: %v", err) + } + adminDB.Close() + }) return db } diff --git a/admin/repository/task.go b/admin/repository/task.go index c336744..6486e45 100644 --- a/admin/repository/task.go +++ b/admin/repository/task.go @@ -1,17 +1,15 @@ package repository import ( + "context" "database/sql" + "errors" "fmt" "strings" "cmautobuy/admin/model" ) -// claimCandidateLimit 是一次最多尝试抢多少条。 -// 抢不到说明被别的客户端拿走了，再试下一条；都抢不到就当作没任务。 -const claimCandidateLimit = 10 - // ClaimNextTask 为指定客户端领取一个任务。 // // 没有可领的任务时返回 (nil, nil) —— 调用方据此返回 204。 @@ -30,13 +28,18 @@ const claimCandidateLimit = 10 // // # 防并发 // -// 做法是**条件更新 + 检查影响行数**：先查出候选，再带着状态条件去更新， -// 影响行数为 0 就说明被别人抢先了，换下一条。 -// 不用 SELECT ... FOR UPDATE，SQLite 没有那个。 +// InnoDB 事务用 FOR UPDATE SKIP LOCKED 锁住一条候选任务。领取状态和 +// task_claims 历史在同一个事务提交，避免只改了状态却没留下领取凭据。 func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model.Task, error) { if clientID == "" { return nil, fmt.Errorf("client_id 不能为空") } + ctx := context.Background() + tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + return nil, fmt.Errorf("开始领取任务事务失败: %w", err) + } + defer tx.Rollback() query := `SELECT task_id FROM tasks WHERE ( (assigned_client = ? AND status = 'assigned') @@ -52,57 +55,45 @@ func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model } } // (assigned_client IS NULL) 为 0/1，0 排前面 —— 指定给本机的优先于无主的 - query += ` ORDER BY (assigned_client IS NULL), priority DESC, created_at LIMIT ?` - args = append(args, claimCandidateLimit) - - rows, err := db.Query(query, args...) - if err != nil { + query += ` ORDER BY (assigned_client IS NULL), priority DESC, created_at + LIMIT 1 FOR UPDATE SKIP LOCKED` + var taskID string + if err := tx.QueryRowContext(ctx, query, args...).Scan(&taskID); errors.Is(err, sql.ErrNoRows) { + return nil, nil + } else if err != nil { return nil, fmt.Errorf("查询可领任务失败: %w", err) } - var candidates []string - for rows.Next() { - var id string - if err := rows.Scan(&id); err != nil { - rows.Close() - return nil, fmt.Errorf("读取候选任务失败: %w", err) - } - candidates = append(candidates, id) - } - rows.Close() - if err := rows.Err(); err != nil { - return nil, err - } now := model.NowISO() - for _, taskID := range candidates { - // 两种情况合成一条语句：对"指定给我的"那种，写 assigned_client - // 是写同一个值，无副作用；对无主的，这一步就是"谁领到就标记谁"。 - res, err := db.Exec(` + // 两种情况合成一条语句：对"指定给我的"那种，写 assigned_client + // 是写同一个值，无副作用；对无主的，这一步就是"谁领到就标记谁"。 + res, err := tx.ExecContext(ctx, ` UPDATE tasks SET status = 'claimed', assigned_client = ?, claimed_at = ?, updated_at = ? WHERE task_id = ? AND ( (status = 'assigned' AND assigned_client = ?) OR (status = 'pending' AND assigned_client IS NULL) )`, - clientID, now, now, taskID, clientID) - if err != nil { - return nil, fmt.Errorf("领取任务 %s 失败: %w", taskID, err) - } - n, err := res.RowsAffected() - if err != nil { - return nil, err - } - if n == 0 { - continue // 被别的客户端抢先了，换下一条 - } - // 记一笔领取历史。提交结果时要靠它判断这台客户端有没有领过—— - // 任务重派后 assigned_client 会变，只看它就查不出来了。 - if err := RecordClaim(db, taskID, clientID, now); err != nil { - return nil, err - } - return GetTask(db, taskID) + clientID, now, now, taskID, clientID) + if err != nil { + return nil, fmt.Errorf("领取任务 %s 失败: %w", taskID, err) } - return nil, nil // 没有可领的任务 + n, err := res.RowsAffected() + if err != nil { + return nil, fmt.Errorf("确认领取任务 %s 结果失败: %w", taskID, err) + } + if n != 1 { + return nil, fmt.Errorf("领取任务 %s 时状态异常", taskID) + } + // 记一笔领取历史。提交结果时要靠它判断这台客户端有没有领过—— + // 任务重派后 assigned_client 会变，只看它就查不出来了。 + if err := RecordClaim(tx, taskID, clientID, now); err != nil { + return nil, err + } + if err := tx.Commit(); err != nil { + return nil, fmt.Errorf("提交领取任务事务失败: %w", err) + } + return GetTask(db, taskID) } // GetTask 按编号读一条任务。 @@ -159,7 +150,7 @@ func RecordClaim(q Execer, taskID, clientID, claimedAt string) error { _, err := q.Exec(` INSERT INTO task_claims (task_id, client_id, claimed_at) VALUES (?, ?, ?) - ON CONFLICT(task_id, client_id) DO UPDATE SET claimed_at = excluded.claimed_at`, + ON DUPLICATE KEY UPDATE claimed_at = VALUES(claimed_at)`, taskID, clientID, claimedAt) if err != nil { return fmt.Errorf("记录领取历史失败 task=%s client=%s: %w", taskID, clientID, err) @@ -312,7 +303,7 @@ func taskFilterClause(filter TaskFilter) (string, []any) { if kw := strings.TrimSpace(filter.Keyword); kw != "" { pattern := "%" + escapeLike(kw) + "%" clauses = append(clauses, - "(t.task_id LIKE ? ESCAPE '\\' OR t.order_no LIKE ? ESCAPE '\\' OR t.pdd_goods_id LIKE ? ESCAPE '\\')") + "(t.task_id LIKE ? ESCAPE '!' OR t.order_no LIKE ? ESCAPE '!' OR t.pdd_goods_id LIKE ? ESCAPE '!')") args = append(args, pattern, pattern, pattern) } diff --git a/admin/repository/user.go b/admin/repository/user.go index 167a5d8..444e429 100644 --- a/admin/repository/user.go +++ b/admin/repository/user.go @@ -5,9 +5,9 @@ import ( "database/sql" "errors" "fmt" - "strings" "cmautobuy/admin/model" + "github.com/go-sql-driver/mysql" ) var ( @@ -63,14 +63,18 @@ func ChangePasswordAndRevokeSessions(db *sql.DB, userID, expectedHash, passwordH return nil } -// CreateInitialAdmin 在写事务里再次确认 users 为空后创建首位管理员。 -// Open 的 _txlock=immediate 让并发事务从一开始就串行，最多一个能成功。 +// CreateInitialAdmin 锁住唯一的初始化哨兵行，再确认 users 为空并创建首位管理员。 +// 并发请求会在 FOR UPDATE 处串行，最多一个能成功。 func CreateInitialAdmin(db *sql.DB, user model.User) error { tx, err := db.Begin() if err != nil { return fmt.Errorf("开始初始化管理员事务失败: %w", err) } defer tx.Rollback() + var lockID int + if err := tx.QueryRow(`SELECT id FROM admin_initialization_lock WHERE id = 1 FOR UPDATE`).Scan(&lockID); err != nil { + return fmt.Errorf("锁定管理员初始化状态失败: %w", err) + } count, err := CountUsers(tx) if err != nil { @@ -189,7 +193,7 @@ func DeleteExpiredSessions(q Execer, nowISO string) error { // ListUsers 按用户名搜索并分页。列表和 COUNT 共用同一段筛选，避免页码漂移。 func ListUsers(q Execer, keyword string, status model.UserStatus, limit, offset int) ([]model.User, int, error) { pattern := "%" + escapeLike(keyword) + "%" - where := `WHERE username LIKE ? ESCAPE '\'` + where := `WHERE username LIKE ? ESCAPE '!'` args := []any{pattern} if status == model.UserActive || status == model.UserDisabled { where += ` AND status = ?` @@ -239,7 +243,8 @@ func CreateUser(q Execer, user model.User) error { user.UserID, user.Username, user.PasswordHash, user.Role, user.Status, user.PasswordChangedAt, user.CreatedAt, user.UpdatedAt) if err != nil { - if strings.Contains(err.Error(), "UNIQUE constraint failed: users.username") { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { return ErrUsernameExists } return fmt.Errorf("创建 Admin 用户失败: %w", err) @@ -258,15 +263,27 @@ func SetUserStatusAndRevokeSessions(db *sql.DB, userID string, status model.User var role model.UserRole var current model.UserStatus - if err := tx.QueryRow(`SELECT role, status FROM users WHERE user_id = ?`, userID).Scan(&role, ¤t); errors.Is(err, sql.ErrNoRows) { + if err := tx.QueryRow(`SELECT role, status FROM users WHERE user_id = ? FOR UPDATE`, userID).Scan(&role, ¤t); errors.Is(err, sql.ErrNoRows) { return ErrUserNotFound } else if err != nil { return fmt.Errorf("读取待修改用户失败: %w", err) } if status == model.UserDisabled && current == model.UserActive && role == model.RoleAdmin { - var activeAdmins int - if err := tx.QueryRow(`SELECT COUNT(*) FROM users WHERE role = 'admin' AND status = 'active'`).Scan(&activeAdmins); err != nil { - return fmt.Errorf("统计有效管理员失败: %w", err) + rows, err := tx.Query(`SELECT user_id FROM users WHERE role = 'admin' AND status = 'active' FOR UPDATE`) + if err != nil { + return fmt.Errorf("锁定有效管理员失败: %w", err) + } + activeAdmins := 0 + for rows.Next() { + var activeID string + if err := rows.Scan(&activeID); err != nil { + rows.Close() + return fmt.Errorf("读取有效管理员失败: %w", err) + } + activeAdmins++ + } + if err := rows.Close(); err != nil { + return fmt.Errorf("读取有效管理员失败: %w", err) } if activeAdmins <= 1 { return ErrLastActiveAdmin diff --git a/admin/service/client_test.go b/admin/service/client_test.go index c0a2552..2c83944 100644 --- a/admin/service/client_test.go +++ b/admin/service/client_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "cmautobuy/admin/internal/testutil" "cmautobuy/admin/model" "cmautobuy/admin/repository" ) @@ -53,16 +54,7 @@ func TestListClientPageForUser_分页统计不泄露其他采购员客户端(t * // 用真实的 migrations，这样表结构变了测试会跟着失败。 func newTestDB(t *testing.T) *sql.DB { t.Helper() - - db, err := repository.Open(t.TempDir()) - if err != nil { - t.Fatalf("打开测试库失败: %v", err) - } - if err := repository.Migrate(db); err != nil { - t.Fatalf("迁移失败: %v", err) - } - t.Cleanup(func() { db.Close() }) - return db + return testutil.OpenMySQL(t) } // insertTask 插一条待领取的任务，供领取相关的测试使用。 diff --git a/admin/service/pdd_page_test.go b/admin/service/pdd_page_test.go index 0aa370f..6be3bc6 100644 --- a/admin/service/pdd_page_test.go +++ b/admin/service/pdd_page_test.go @@ -305,7 +305,7 @@ func TestListPddProducts_规格数(t *testing.T) { } // skus_json 存进了坏数据时，列表页必须还能打开。 -// json_array_length 碰到非法 JSON 会让整条查询报错，那样页面直接白屏。 +// JSON_LENGTH 碰到非法 JSON 会让整条查询报错，那样页面直接白屏。 func TestListPddProducts_坏掉的采集结果不影响列表打开(t *testing.T) { db := newTestDB(t) createProduct(t, db, "100000000001") @@ -843,7 +843,7 @@ func insertStaleCollectingProduct(t *testing.T, db *sql.DB, goodsID string) { INSERT INTO tasks (task_id, task_type, status, assigned_client, pdd_goods_url, pdd_goods_id, created_at, updated_at) VALUES (?, 'collect', 'claimed', 'client-已离线', - 'https://mobile.yangkeduo.com/goods.html?goods_id=' || ?, ?, ?, ?)`, + CONCAT('https://mobile.yangkeduo.com/goods.html?goods_id=', ?), ?, ?, ?)`, "COL-STALE-"+goodsID, goodsID, goodsID, staleAt, staleAt); err != nil { t.Fatalf("构造卡死任务失败: %v", err) } diff --git a/admin/service/pdd_test.go b/admin/service/pdd_test.go index 4c2be78..7d75fcd 100644 --- a/admin/service/pdd_test.go +++ b/admin/service/pdd_test.go @@ -206,7 +206,7 @@ func insertShopeeSKU(t *testing.T, db *sql.DB, goodsID, skuID string) { if _, err := db.Exec(` INSERT INTO shopee_products (goods_id, title, created_at, updated_at) VALUES (?, '测试商品', ?, ?) - ON CONFLICT(goods_id) DO NOTHING`, goodsID, now, now); err != nil { + ON DUPLICATE KEY UPDATE goods_id = VALUES(goods_id)`, goodsID, now, now); err != nil { t.Fatalf("插入蝦皮商品失败: %v", err) } if _, err := db.Exec(` diff --git a/admin/service/purchase_workflow_test.go b/admin/service/purchase_workflow_test.go index 4432daf..43f81cf 100644 --- a/admin/service/purchase_workflow_test.go +++ b/admin/service/purchase_workflow_test.go @@ -57,11 +57,17 @@ func TestSybMapping_动态维度保存并复用(t *testing.T) { if err := SaveSybMapping(db, "SYB-1", key, "USR-1"); err != nil { t.Fatal(err) } - detail, _ = GetSybProcessingDetail(db, "SYB-1") + detail, err = GetSybProcessingDetail(db, "SYB-1") + if err != nil { + t.Fatal(err) + } if !detail.MappingValid { t.Fatal("保存后相同蝦皮 SKU + 当前 PDD 商品应自动复用") } - list, _ := ListSybOrdersView(db, "", SybStagePurchaseReady, 1) + list, err := ListSybOrdersView(db, "", SybStagePurchaseReady, 1) + if err != nil { + t.Fatal(err) + } if list.Total != 1 || !list.Rows[0].CanPurchase || list.Rows[0].DefaultMaxPrice != "39.90" { t.Fatalf("可采购阶段不对: %+v", list) } diff --git a/admin/service/submit.go b/admin/service/submit.go index aee6a4d..1c06d55 100644 --- a/admin/service/submit.go +++ b/admin/service/submit.go @@ -292,6 +292,22 @@ func submitInTx( // 5. 记下幂等键。和业务写入在同一个事务里—— // 分开写的话，业务写成功但幂等没记上，客户端重试会被重复处理。 if err := repository.SaveIdempotent(tx, idemKey, hash, string(respJSON)); err != nil { + // 两个首次请求可能同时通过前面的“不存在”检查。唯一键会让其中一个 + // 等待另一个提交后返回重复键；此时回滚本事务的业务写入，再读取赢家 + // 保存的响应，保证并发重试拿到完全相同的结果。 + if errors.Is(err, repository.ErrIdempotencyAlreadySaved) { + if rollbackErr := tx.Rollback(); rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) { + return "", fmt.Errorf("回滚重复幂等请求失败: %w", rollbackErr) + } + body, done, lookupErr := repository.LookupIdempotent(db, idemKey, hash) + if lookupErr != nil { + return "", lookupErr + } + if !done { + return "", fmt.Errorf("并发幂等请求已提交但未找到保存结果") + } + return body, nil + } return "", err } diff --git a/admin/service/syb_test.go b/admin/service/syb_test.go index 4938fc4..ca519ca 100644 --- a/admin/service/syb_test.go +++ b/admin/service/syb_test.go @@ -359,15 +359,7 @@ func writeEnvelope(t *testing.T, w http.ResponseWriter, status bool, msg string, func newSyncTestDB(t *testing.T) *sql.DB { t.Helper() - db, err := repository.Open(t.TempDir()) - if err != nil { - t.Fatalf("打开测试库失败: %v", err) - } - t.Cleanup(func() { db.Close() }) - if err := repository.Migrate(db); err != nil { - t.Fatalf("迁移失败: %v", err) - } - return db + return newTestDB(t) } func TestRunSybSync_已有的ShopeeSKUID同步后仍在(t *testing.T) { diff --git a/admin/user_integration_test.go b/admin/user_integration_test.go index c35210a..c3c3560 100644 --- a/admin/user_integration_test.go +++ b/admin/user_integration_test.go @@ -13,14 +13,7 @@ import ( ) func Test用户管理管理员与采购员权限边界(t *testing.T) { - db, err := repository.Open(t.TempDir()) - if err != nil { - t.Fatal(err) - } - defer db.Close() - if err := repository.Migrate(db); err != nil { - t.Fatal(err) - } + db := newMySQLTestDB(t) now := time.Now() if err := service.SetupInitialAdmin(db, "admin", "admin-password", "admin-password", now); err != nil { t.Fatal(err) diff --git a/docs/admin/00-glossary.md b/docs/admin/00-glossary.md index ceef7c6..de902d8 100644 --- a/docs/admin/00-glossary.md +++ b/docs/admin/00-glossary.md @@ -29,8 +29,8 @@ | 服务端渲染 | 页面的 HTML 在服务器上拼好再发给浏览器 | 与之相对的是前端框架在浏览器里拼，本项目**不用** | | htmx | 一个单文件 JS 库，让 HTML 标签直接发请求换局部内容 | 需要局部刷新时可用，**没有构建步骤** | | upsert | "有就更新、没有就新增"，一次操作搞定 | Excel 导入的唯一正确做法，见 §3 | -| cgo | Go 调用 C 代码的机制。**用了就需要装 C 编译器** | 本项目**避开它**，所以 SQLite 驱动选纯 Go 的 | -| `modernc.org/sqlite` | 纯 Go 实现的 SQLite，不需要 cgo | 固定用它，`go build` 直接出 exe | +| cgo | Go 调用 C 代码的机制。**用了就需要装 C 编译器** | 本项目**避开它**；MySQL 和历史 SQLite 迁移驱动都是纯 Go | +| `modernc.org/sqlite` | 纯 Go 实现的 SQLite，不需要 cgo | 只用于读取历史 `admin.db` 和迁移回归，不进入生产运行时 | | excelize | Go 读写 Excel 的库 | 固定用它读蝦皮报表 | | CSRF | 攻击者诱导你在已登录状态下发出非本意的请求 | 所有写操作都要防，见 [06](06-quality-security.md) §4 | | 参数化查询 | SQL 里用 `?` 占位、值单独传，而不是拼字符串 | 防 SQL 注入的唯一正确做法 | diff --git a/docs/admin/05-ui-specification.md b/docs/admin/05-ui-specification.md index 9dd9110..0f841c5 100644 --- a/docs/admin/05-ui-specification.md +++ b/docs/admin/05-ui-specification.md @@ -271,9 +271,9 @@ M 个 SKU，K 行解析失败"），覆盖掉正常的统计文案，并且回 前者是还没采，后者是采集出了问题（商品下架、页面改版、解析器没认出来）， 都显示成 `—` 就看不出区别了。 -`[建议]` 规格数用 SQLite 的 `json_array_length(skus_json, '$.skus')` 直接算， -不要把整个 JSON 读进 Go 再数。`[必须]` 外面要包一层 `json_valid`—— -`skus_json` 万一存进了坏数据，`json_array_length` 会让**整条查询报错**，页面直接打不开。 +`[建议]` 规格数用 MySQL 的 `JSON_LENGTH(skus_json, '$.skus')` 直接算， +不要把整个 JSON 读进 Go 再数。`[必须]` 外面要包一层 `JSON_VALID`—— +`skus_json` 万一存进了坏数据，`JSON_LENGTH` 会让**整条查询报错**，页面直接打不开。 `[必须]` 空状态分两种文案：从没创建过 → 引导去点「创建」；筛选无结果 → 给"查看全部"的入口。 diff --git a/docs/admin/08-顺运宝接口.md b/docs/admin/08-顺运宝接口.md index 3837aec..323984c 100644 --- a/docs/admin/08-顺运宝接口.md +++ b/docs/admin/08-顺运宝接口.md @@ -412,9 +412,9 @@ Go 侧不用跟着调。 这几条不是抓包结论，是本项目的决定，写在这里避免每次重新讨论： -`[必须]` **会话缓存存 SQLite，不引入 Redis。** Admin 的定位是「双击 exe 就能跑」， -`data/` 在 exe 旁边。示例脚本用 Redis 是因为它是反复启动的一次性脚本， -进程间要传会话；Admin 是常驻进程，没有这个需求，持久化只为重启后免登录。 +`[必须]` **会话缓存存 Admin 的 MySQL，不引入 Redis。** 示例脚本用 Redis 是因为 +它是反复启动的一次性脚本，进程间要传会话；Admin 是常驻进程，没有这个需求， +持久化只为重启后免登录，继续复用现有数据库即可。 `[决定已变更]` ~~不引入 OCR 服务。~~ 这条判断在工单 #47 里被推翻了， 原文和推翻理由都留在这里，方便后来人知道这个决定变过、为什么变：