package repository import ( "context" "crypto/tls" "crypto/x509" "database/sql" "errors" "fmt" "log" "net" "os" "strings" "time" "github.com/go-sql-driver/mysql" "cmautobuy/admin/config" "cmautobuy/admin/model" "cmautobuy/admin/spec" ) const mysqlSchemaVersion = 24 // OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。 func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) { driverConfig, err := newMySQLDriverConfig(cfg) if err != nil { return nil, err } connector, err := mysql.NewConnector(driverConfig) if err != nil { return nil, fmt.Errorf("准备 MySQL 连接失败: %w", err) } db := sql.OpenDB(connector) db.SetMaxOpenConns(10) db.SetMaxIdleConns(10) db.SetConnMaxLifetime(3 * time.Minute) db.SetConnMaxIdleTime(time.Minute) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := db.PingContext(ctx); err != nil { db.Close() return nil, fmt.Errorf("连接 MySQL 失败，请检查服务、库名、config.yaml、环境变量和 TLS 配置: %w", err) } if err := CheckMySQLServer(db, cfg.Name); err != nil { db.Close() return nil, err } return db, nil } // newMySQLDriverConfig 只负责把业务配置转换成驱动配置，方便单元测试在不连接 // 数据库的情况下核对 TLS 是否真的开启。 func newMySQLDriverConfig(cfg config.DatabaseConfig) (*mysql.Config, error) { driverConfig := mysql.NewConfig() driverConfig.User = cfg.User driverConfig.Passwd = cfg.Password driverConfig.Net = "tcp" driverConfig.Addr = net.JoinHostPort(cfg.Host, cfg.Port) driverConfig.DBName = cfg.Name driverConfig.Collation = "utf8mb4_0900_ai_ci" driverConfig.Loc = time.UTC driverConfig.Timeout = 5 * time.Second 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'", } if cfg.TLSMode == config.DatabaseTLSVerifyCA { tlsConfig, err := loadMySQLTLSConfig(cfg.TLSCA) if err != nil { return nil, err } driverConfig.TLS = tlsConfig } return driverConfig, nil } // loadMySQLTLSConfig 加载服务器专属 CA，并验证服务端证书确实由它签发。 // MySQL 自动生成的服务端证书没有 SAN，Go 无法做 IP/域名匹配；这里显式跳过 // 内置主机名检查，但用 VerifyConnection 恢复证书链校验，不能退回明文连接。 func loadMySQLTLSConfig(caPath string) (*tls.Config, error) { pemData, err := os.ReadFile(caPath) if err != nil { return nil, fmt.Errorf("读取 MySQL TLS CA 文件 %s 失败: %w", caPath, err) } roots := x509.NewCertPool() if !roots.AppendCertsFromPEM(pemData) { return nil, fmt.Errorf("解析 MySQL TLS CA 文件 %s 失败: 文件中没有有效 PEM 证书", caPath) } return &tls.Config{ MinVersion: tls.VersionTLS12, InsecureSkipVerify: true, // 主机名检查由下面的服务器专属 CA 链校验替代。 VerifyConnection: func(state tls.ConnectionState) error { if len(state.PeerCertificates) == 0 { return fmt.Errorf("验证 MySQL TLS 证书失败: 服务端没有提供证书") } intermediates := x509.NewCertPool() for _, certificate := range state.PeerCertificates[1:] { intermediates.AddCert(certificate) } _, err := state.PeerCertificates[0].Verify(x509.VerifyOptions{ Roots: roots, Intermediates: intermediates, KeyUsages: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, }) if err != nil { return fmt.Errorf("验证 MySQL TLS 证书链失败: %w", err) } return nil }, }, nil } // CheckMySQLServer 拒绝错误版本、错误库、非 UTC 或非 utf8mb4 的连接。 func CheckMySQLServer(db *sql.DB, expectedDatabase string) error { var version, databaseName, timeZone, charset, collation string if err := db.QueryRow(`SELECT VERSION(), DATABASE(), @@session.time_zone, @@character_set_connection, @@collation_connection`).Scan( &version, &databaseName, &timeZone, &charset, &collation); err != nil { return fmt.Errorf("读取 MySQL 运行参数失败: %w", err) } if !strings.HasPrefix(version, "8.") { return fmt.Errorf("MySQL 版本不兼容：需要 8.x，实际 %s", version) } if databaseName != expectedDatabase { return fmt.Errorf("连接到了错误的 MySQL 数据库：期望 %s，实际 %s", expectedDatabase, databaseName) } if timeZone != "+00:00" { return fmt.Errorf("MySQL 会话时区必须是 +00:00，实际 %s", timeZone) } if charset != "utf8mb4" || !strings.HasPrefix(collation, "utf8mb4_") { return fmt.Errorf("MySQL 连接字符集必须是 utf8mb4，实际 %s/%s", charset, collation) } return nil } // mysqlSchemaV1 是当前 SQLite v8 最终业务结构在 MySQL 8 上的等价定义。 // 每条 CREATE TABLE 都可重复执行；全部完成并通过自检后才记录版本。 var mysqlSchemaV1 = []string{ `CREATE TABLE IF NOT EXISTS shopee_products ( goods_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, title TEXT NOT NULL, shopee_status VARCHAR(191), main_sku_code VARCHAR(191), pdd_goods_url TEXT, pdd_goods_id VARCHAR(191) COLLATE utf8mb4_bin, created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL, KEY idx_shopee_products_pdd (pdd_goods_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS shopee_skus ( sku_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, spec_raw TEXT NOT NULL, color TEXT, size TEXT, advice TEXT, parse_ok TINYINT NOT NULL DEFAULT 0, sku_code VARCHAR(191), is_manual TINYINT NOT NULL DEFAULT 0, created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL, KEY idx_shopee_skus_goods (goods_id), KEY idx_shopee_skus_parse (parse_ok), CONSTRAINT fk_shopee_skus_product FOREIGN KEY (goods_id) REFERENCES shopee_products(goods_id) ON DELETE CASCADE, CHECK (parse_ok IN (0, 1)), CHECK (is_manual IN (0, 1)) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS pdd_products ( id BIGINT PRIMARY KEY AUTO_INCREMENT, goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL UNIQUE, url TEXT NOT NULL, title TEXT, shop_name TEXT, skus_json LONGTEXT, collect_status VARCHAR(20) NOT NULL DEFAULT 'pending', collect_msg TEXT, artifact_ref TEXT, collected_at VARCHAR(35), deleted_at VARCHAR(35), created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL, KEY idx_pdd_products_status (collect_status), CHECK (collect_status IN ('pending', 'collecting', 'collected', 'failed')) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS syb_orders ( syb_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, order_no VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, title TEXT, shopee_goods_id VARCHAR(191) COLLATE utf8mb4_bin, shopee_sku_id VARCHAR(191) COLLATE utf8mb4_bin, product_spec TEXT, quantity BIGINT NOT NULL, price_twd_cent BIGINT, image_url TEXT, syb_data LONGTEXT NOT NULL DEFAULT ('{}'), created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL, KEY idx_syb_orders_order (order_no), KEY idx_syb_orders_goods (shopee_goods_id), KEY idx_syb_orders_list (updated_at DESC, syb_id DESC), CHECK (quantity > 0), CHECK (price_twd_cent IS NULL OR price_twd_cent >= 0) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS sku_mappings ( shopee_sku_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, pdd_goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, pdd_option_key VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, pdd_options LONGTEXT NOT NULL, goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, mapped_at VARCHAR(35) NOT NULL, mapped_by VARCHAR(191), PRIMARY KEY (shopee_sku_id, pdd_goods_id), KEY idx_sku_mappings_goods (goods_id), KEY idx_sku_mappings_pdd (pdd_goods_id), CONSTRAINT fk_sku_mappings_sku FOREIGN KEY (shopee_sku_id) REFERENCES shopee_skus(sku_id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS tasks ( task_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, task_type VARCHAR(20) NOT NULL, status VARCHAR(20) NOT NULL DEFAULT 'pending', version BIGINT NOT NULL DEFAULT 1, priority BIGINT NOT NULL DEFAULT 0, assigned_client VARCHAR(191) COLLATE utf8mb4_bin, claimed_at VARCHAR(35), syb_id VARCHAR(191) COLLATE utf8mb4_bin, order_no VARCHAR(191) COLLATE utf8mb4_bin, goods_id VARCHAR(191) COLLATE utf8mb4_bin, shopee_sku_id VARCHAR(191) COLLATE utf8mb4_bin, pdd_goods_url TEXT NOT NULL, pdd_goods_id VARCHAR(191) COLLATE utf8mb4_bin, pdd_options LONGTEXT, quantity BIGINT, max_price_cent BIGINT, result_data LONGTEXT, error_code VARCHAR(191), error_message TEXT, finished_at VARCHAR(35), created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL, KEY idx_tasks_claim (assigned_client, status, priority DESC, created_at), KEY idx_tasks_list (updated_at DESC, task_id DESC), KEY idx_tasks_order (order_no), CHECK (task_type IN ('collect', 'purchase')), CHECK (status IN ('pending', 'assigned', 'claimed', 'succeeded', 'manual_review', 'failed', 'cancelled')), CHECK (version > 0), CHECK (quantity IS NULL OR quantity > 0), CHECK (max_price_cent IS NULL OR max_price_cent > 0) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS clients ( client_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, name TEXT, device_address TEXT, platform VARCHAR(191), pdd_package VARCHAR(191), capabilities LONGTEXT, last_seen_at VARCHAR(35) NOT NULL, created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS idempotency_keys ( ` + "`key`" + ` VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, request_hash VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, response_body LONGTEXT NOT NULL, created_at VARCHAR(35) NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS task_claims ( task_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, client_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, claimed_at VARCHAR(35) NOT NULL, PRIMARY KEY (task_id, client_id), KEY idx_task_claims_client (client_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS syb_session ( username VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, cookies LONGTEXT NOT NULL, expires_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS syb_sync_state ( id TINYINT PRIMARY KEY, last_synced_at VARCHAR(35), updated_at VARCHAR(35) NOT NULL, CHECK (id = 1) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS users ( user_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, username VARCHAR(191) COLLATE utf8mb4_0900_ai_ci NOT NULL UNIQUE, password_hash VARCHAR(255) COLLATE utf8mb4_bin NOT NULL, role VARCHAR(20) NOT NULL, status VARCHAR(20) NOT NULL, last_login_at VARCHAR(35), password_changed_at VARCHAR(35) NOT NULL, created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL, CHECK (role IN ('admin', 'purchaser')), CHECK (status IN ('active', 'disabled')) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS web_sessions ( session_hash VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, expires_at VARCHAR(35) NOT NULL, created_at VARCHAR(35) NOT NULL, last_seen_at VARCHAR(35) NOT NULL, KEY idx_web_sessions_user (user_id), KEY idx_web_sessions_expiry (expires_at), CONSTRAINT fk_web_sessions_user FOREIGN KEY (user_id) REFERENCES users(user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS client_user_assignments ( assignment_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, client_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, started_at VARCHAR(35) NOT NULL, ended_at VARCHAR(35), assigned_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, ended_by_user_id VARCHAR(191) COLLATE utf8mb4_bin, end_reason VARCHAR(20), current_client_id VARCHAR(191) COLLATE utf8mb4_bin GENERATED ALWAYS AS (IF(ended_at IS NULL, client_id, NULL)) STORED, UNIQUE KEY idx_client_assignment_current (current_client_id), KEY idx_client_assignment_user (user_id, ended_at, client_id), KEY idx_client_assignment_history (client_id, started_at DESC), CONSTRAINT fk_client_assignment_user FOREIGN KEY (user_id) REFERENCES users(user_id), CONSTRAINT fk_client_assignment_assigned_by FOREIGN KEY (assigned_by_user_id) REFERENCES users(user_id), CONSTRAINT fk_client_assignment_ended_by FOREIGN KEY (ended_by_user_id) REFERENCES users(user_id), CHECK (end_reason IS NULL OR end_reason IN ('unbind', 'transfer')), CHECK ((ended_at IS NULL AND ended_by_user_id IS NULL AND end_reason IS NULL) OR (ended_at IS NOT NULL AND ended_by_user_id IS NOT NULL AND end_reason IS NOT NULL)) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS syb_sync_runs ( run_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, date_from VARCHAR(10) NOT NULL, date_to VARCHAR(10) NOT NULL, status VARCHAR(20) NOT NULL, stock_count BIGINT NOT NULL DEFAULT 0, detail_count BIGINT NOT NULL DEFAULT 0, created_count BIGINT NOT NULL DEFAULT 0, updated_count BIGINT NOT NULL DEFAULT 0, skipped_count BIGINT NOT NULL DEFAULT 0, error_message TEXT, cursor_advanced TINYINT NOT NULL DEFAULT 0, started_at VARCHAR(35) NOT NULL, finished_at VARCHAR(35), KEY idx_syb_sync_runs_started (started_at DESC, run_id DESC), CONSTRAINT fk_syb_sync_runs_user FOREIGN KEY (user_id) REFERENCES users(user_id), CHECK (status IN ('running', 'succeeded', 'failed', 'interrupted')), CHECK (cursor_advanced IN (0, 1)), CHECK ((status = 'running' AND finished_at IS NULL) OR (status <> 'running' AND finished_at IS NOT NULL)) ) 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)`, } const mysqlSchemaV3SpecMappings = `CREATE TABLE IF NOT EXISTS spec_mappings ( shopee_goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, spec_key VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, pdd_goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, pdd_option_key VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, pdd_options LONGTEXT NOT NULL, spec_raw TEXT NOT NULL, mapped_at VARCHAR(35) NOT NULL, mapped_by VARCHAR(191), PRIMARY KEY (shopee_goods_id, spec_key, pdd_goods_id), KEY idx_spec_mappings_goods (shopee_goods_id), KEY idx_spec_mappings_pdd (pdd_goods_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci` const mysqlSchemaV4Decisions = `CREATE TABLE IF NOT EXISTS spec_mapping_decisions ( id BIGINT AUTO_INCREMENT PRIMARY KEY, shopee_goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, spec_key VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, pdd_goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, rules_version VARCHAR(32) NOT NULL, suggested_option_key VARCHAR(191) COLLATE utf8mb4_bin, chosen_option_key VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, accepted TINYINT NOT NULL, decided_by VARCHAR(191), decided_at VARCHAR(35) NOT NULL, CONSTRAINT chk_spec_mapping_decisions_accepted CHECK (accepted IN (0,1)), KEY idx_decisions_mapping (shopee_goods_id, spec_key, pdd_goods_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci` // MigrateMySQL 建立或升级 MySQL schema。生产迁移只能在这里追加新版本。 func MigrateMySQL(db *sql.DB) error { if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations ( version INT PRIMARY KEY, applied_at VARCHAR(35) NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`); err != nil { return fmt.Errorf("建立 MySQL 迁移版本表失败: %w", err) } var current int if err := db.QueryRow(`SELECT COALESCE(MAX(version), 0) FROM schema_migrations`).Scan(¤t); err != nil { return fmt.Errorf("读取 MySQL schema 版本失败: %w", err) } if current > mysqlSchemaVersion { return fmt.Errorf("MySQL schema 版本 %d 高于程序支持的 %d，请升级程序", current, mysqlSchemaVersion) } if current < 1 { for i, statement := range mysqlSchemaV1 { if _, err := db.Exec(statement); err != nil { return fmt.Errorf("执行 MySQL schema v1 第 %d 条失败: %w", i+1, err) } } 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) } } mysqlRequiredTablesV2 := append(append([]string{}, requiredTables...), "admin_initialization_lock") if err := checkMySQLSchema(db, mysqlRequiredTablesV2); 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) } current = 2 } if current < 3 { if err := migrateMySQLV3(db); err != nil { return fmt.Errorf("执行 MySQL schema v3 失败: %w", err) } if err := checkMySQLSchemaV3(db); err != nil { return fmt.Errorf("MySQL schema v3 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 3, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v3 失败: %w", err) } current = 3 } if current < 4 { if _, err := db.Exec(mysqlSchemaV4Decisions); err != nil { return fmt.Errorf("执行 MySQL schema v4 失败: %w", err) } if err := checkMySQLV4Shape(db); err != nil { return fmt.Errorf("MySQL schema v4 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 4, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v4 失败: %w", err) } current = 4 } if current < 5 { if err := migrateMySQLV5(db); err != nil { return fmt.Errorf("执行 MySQL schema v5 失败: %w", err) } if err := checkMySQLV5Shape(db); err != nil { return fmt.Errorf("MySQL schema v5 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 5, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v5 失败: %w", err) } current = 5 } if current < 6 { if err := migrateMySQLV6(db); err != nil { return fmt.Errorf("执行 MySQL schema v6 失败: %w", err) } if err := checkMySQLV6Shape(db); err != nil { return fmt.Errorf("MySQL schema v6 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 6, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v6 失败: %w", err) } current = 6 } if current < 7 { if err := migrateMySQLV7(db); err != nil { return fmt.Errorf("执行 MySQL schema v7 失败: %w", err) } if err := checkMySQLV7Shape(db); err != nil { return fmt.Errorf("MySQL schema v7 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 7, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v7 失败: %w", err) } current = 7 } if current < 8 { if err := migrateMySQLV8(db); err != nil { return fmt.Errorf("执行 MySQL schema v8 失败: %w", err) } if err := checkMySQLV8Shape(db); err != nil { return fmt.Errorf("MySQL schema v8 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 8, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v8 失败: %w", err) } current = 8 } if current < 9 { if err := migrateMySQLV9(db); err != nil { return fmt.Errorf("执行 MySQL schema v9 失败: %w", err) } if err := checkMySQLV9Shape(db); err != nil { return fmt.Errorf("MySQL schema v9 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 9, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v9 失败: %w", err) } current = 9 } if current < 10 { if err := migrateMySQLV10(db); err != nil { return fmt.Errorf("执行 MySQL schema v10 失败: %w", err) } if err := checkMySQLV10Shape(db); err != nil { return fmt.Errorf("MySQL schema v10 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 10, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v10 失败: %w", err) } current = 10 } if current < 11 { if err := migrateMySQLV11(db); err != nil { return fmt.Errorf("执行 MySQL schema v11 失败: %w", err) } if err := checkMySQLV11Shape(db); err != nil { return fmt.Errorf("MySQL schema v11 自检失败，未记录版本: %w", err) } 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) } 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) } current = 13 } if current < 14 { if err := migrateMySQLV14(db); err != nil { return fmt.Errorf("执行 MySQL schema v14 失败: %w", err) } if err := checkMySQLV14Shape(db); err != nil { return fmt.Errorf("MySQL schema v14 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 14, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v14 失败: %w", err) } current = 14 } if current < 15 { if err := migrateMySQLV15(db); err != nil { return fmt.Errorf("执行 MySQL schema v15 失败: %w", err) } if err := checkMySQLV15Shape(db); err != nil { return fmt.Errorf("MySQL schema v15 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 15, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v15 失败: %w", err) } current = 15 } if current < 16 { if err := migrateMySQLV16(db); err != nil { return fmt.Errorf("执行 MySQL schema v16 失败: %w", err) } if err := checkMySQLV16Shape(db); err != nil { return fmt.Errorf("MySQL schema v16 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 16, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v16 失败: %w", err) } current = 16 } if current < 17 { if err := migrateMySQLV17(db); err != nil { return fmt.Errorf("执行 MySQL schema v17 失败: %w", err) } if err := checkMySQLV17Shape(db); err != nil { return fmt.Errorf("MySQL schema v17 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 17, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v17 失败: %w", err) } current = 17 } if current < 18 { if err := migrateMySQLV18(db); err != nil { return fmt.Errorf("执行 MySQL schema v18 失败: %w", err) } if err := checkMySQLV18Shape(db); err != nil { return fmt.Errorf("MySQL schema v18 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 18, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v18 失败: %w", err) } current = 18 } if current < 19 { if err := migrateMySQLV19(db); err != nil { return fmt.Errorf("执行 MySQL schema v19 失败: %w", err) } if err := checkMySQLV19Shape(db); err != nil { return fmt.Errorf("MySQL schema v19 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 19, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v19 失败: %w", err) } current = 19 } if current < 20 { if err := migrateMySQLV20(db); err != nil { return fmt.Errorf("执行 MySQL schema v20 失败: %w", err) } if err := checkMySQLV20Shape(db); err != nil { return fmt.Errorf("MySQL schema v20 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 20, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v20 失败: %w", err) } current = 20 } if current < 21 { if err := migrateMySQLV21(db); err != nil { return fmt.Errorf("执行 MySQL schema v21 失败: %w", err) } if err := checkMySQLV21Shape(db); err != nil { return fmt.Errorf("MySQL schema v21 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 21, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v21 失败: %w", err) } current = 21 } if current < 22 { if err := migrateMySQLV22(db); err != nil { return fmt.Errorf("执行 MySQL schema v22 失败: %w", err) } if err := checkMySQLV22Shape(db); err != nil { return fmt.Errorf("MySQL schema v22 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 22, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v22 失败: %w", err) } current = 22 } if current < 23 { if err := migrateMySQLV23(db); err != nil { return fmt.Errorf("执行 MySQL schema v23 失败: %w", err) } if err := checkMySQLV23Shape(db); err != nil { return fmt.Errorf("MySQL schema v23 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 23, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v23 失败: %w", err) } current = 23 } if current < 24 { if err := migrateMySQLV24(db); err != nil { return fmt.Errorf("执行 MySQL schema v24 失败: %w", err) } if err := checkMySQLV24Shape(db); err != nil { return fmt.Errorf("MySQL schema v24 自检失败，未记录版本: %w", err) } if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 24, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v24 失败: %w", err) } } return CheckMySQLSchema(db) } // migrateMySQLV24 为档口入库码增加可恢复软删除审计。 func migrateMySQLV24(db *sql.DB) error { for _, column := range []struct{ name, ddl string }{ {"deleted_at", `ALTER TABLE syb_inner_code_records ADD COLUMN deleted_at VARCHAR(35) NULL AFTER applied_at`}, {"deleted_by_user_id", `ALTER TABLE syb_inner_code_records ADD COLUMN deleted_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER deleted_at`}, } { exists, err := mysqlColumnExists(db, "syb_inner_code_records", column.name) if err != nil { return err } if !exists { if _, err := db.Exec(column.ddl); err != nil { return fmt.Errorf("增加 syb_inner_code_records.%s 失败: %w", column.name, err) } } } if exists, err := mysqlIndexExists(db, "syb_inner_code_records", "idx_inner_code_deleted"); err != nil { return err } else if !exists { if _, err := db.Exec(`ALTER TABLE syb_inner_code_records ADD INDEX idx_inner_code_deleted (deleted_at,business_date,status,id)`); err != nil { return fmt.Errorf("增加档口入库码软删除索引失败: %w", err) } } if exists, err := mysqlConstraintExists(db, "syb_inner_code_records", "fk_inner_code_deleted_by"); err != nil { return err } else if !exists { if _, err := db.Exec(`ALTER TABLE syb_inner_code_records ADD CONSTRAINT fk_inner_code_deleted_by FOREIGN KEY (deleted_by_user_id) REFERENCES users(user_id) ON DELETE SET NULL`); err != nil { return fmt.Errorf("增加档口入库码删除人外键失败: %w", err) } } return nil } // migrateMySQLV23 建立档口入库码最小闭环使用的唯一业务表。 // Excel 原文件、批次和每次尝试不会另建表保存。 func migrateMySQLV23(db *sql.DB) error { _, err := db.Exec(`CREATE TABLE IF NOT EXISTS syb_inner_code_records ( id BIGINT AUTO_INCREMENT PRIMARY KEY, business_date DATE NOT NULL, source_row INT NOT NULL, print_sequence INT NULL, order_number VARCHAR(64) COLLATE utf8mb4_bin NOT NULL, shop_name VARCHAR(191) NULL, stall VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, spec_raw VARCHAR(500) NOT NULL, spec_key VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, inner_code VARCHAR(128) COLLATE utf8mb4_bin NOT NULL, source_duplicate_count INT NOT NULL DEFAULT 1, stock_id BIGINT NULL, detail_id BIGINT NULL, syb_spec VARCHAR(500) NULL, syb_sku VARCHAR(191) NULL, syb_variation_sku VARCHAR(191) NULL, purchase_platform VARCHAR(191) NULL, purchase_code VARCHAR(191) NULL, remote_inner_code VARCHAR(128) NULL, status VARCHAR(32) COLLATE utf8mb4_bin NOT NULL DEFAULT 'pending', result_message VARCHAR(500) NULL, created_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, applied_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NULL, planned_at VARCHAR(35) NULL, apply_started_at VARCHAR(35) NULL, applied_at VARCHAR(35) NULL, created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL, UNIQUE KEY uq_inner_code_business (business_date,order_number,stall,spec_key), UNIQUE KEY uq_inner_code_date_code (business_date,inner_code), KEY idx_inner_code_list (business_date,status,id DESC), KEY idx_inner_code_order (order_number,business_date,id), KEY idx_inner_code_detail (detail_id,status,id), CONSTRAINT fk_inner_code_created_by FOREIGN KEY (created_by_user_id) REFERENCES users(user_id), CONSTRAINT fk_inner_code_applied_by FOREIGN KEY (applied_by_user_id) REFERENCES users(user_id), CONSTRAINT chk_inner_code_status CHECK (status IN ('pending','ready','applying','updated','already_filled','skipped','failed','needs_check')), CONSTRAINT chk_inner_code_source_row CHECK (source_row > 0), CONSTRAINT chk_inner_code_duplicate_count CHECK (source_duplicate_count > 0) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`) return err } func migrateMySQLV22(db *sql.DB) error { statements := []string{ `CREATE TABLE IF NOT EXISTS ai_match_batches ( batch_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, status VARCHAR(16) NOT NULL, created_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, total_count INT NOT NULL,processed_count INT NOT NULL DEFAULT 0, success_count INT NOT NULL DEFAULT 0,reused_count INT NOT NULL DEFAULT 0, manual_count INT NOT NULL DEFAULT 0,failed_count INT NOT NULL DEFAULT 0, provider_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, provider_name VARCHAR(191) NOT NULL,provider_base_url VARCHAR(2048) NOT NULL, model VARCHAR(191) NOT NULL,timeout_seconds INT NOT NULL,max_concurrency INT NOT NULL, confidence_threshold_bps INT NOT NULL,config_fingerprint CHAR(64) COLLATE utf8mb4_bin NOT NULL, rules_version VARCHAR(32) NOT NULL,prompt_version VARCHAR(32) NOT NULL, error_message VARCHAR(500),created_at VARCHAR(35) NOT NULL, started_at VARCHAR(35),finished_at VARCHAR(35), KEY idx_ai_match_batch_user (created_by_user_id,created_at DESC,batch_id), KEY idx_ai_match_batch_status (status,created_at,batch_id), CONSTRAINT fk_ai_match_batch_user FOREIGN KEY (created_by_user_id) REFERENCES users(user_id), CONSTRAINT chk_ai_match_batch_status CHECK (status IN ('queued','running','partial','succeeded','failed','interrupted')), CONSTRAINT chk_ai_match_batch_total CHECK (total_count BETWEEN 1 AND 100), CONSTRAINT chk_ai_match_batch_counts CHECK (processed_count BETWEEN 0 AND total_count AND success_count>=0 AND reused_count>=0 AND manual_count>=0 AND failed_count>=0), CONSTRAINT chk_ai_match_batch_runtime CHECK (timeout_seconds BETWEEN 1 AND 120 AND max_concurrency BETWEEN 1 AND 16 AND confidence_threshold_bps BETWEEN 0 AND 10000) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS ai_match_batch_items ( item_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, batch_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, syb_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, identity_hash CHAR(64) COLLATE utf8mb4_bin NOT NULL, leader_syb_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, context_version CHAR(64) COLLATE utf8mb4_bin NOT NULL, position INT NOT NULL,status VARCHAR(16) NOT NULL,outcome VARCHAR(32), message VARCHAR(500),mapping_source VARCHAR(16),option_key VARCHAR(191) COLLATE utf8mb4_bin, confidence_bps INT,model_called TINYINT NOT NULL DEFAULT 0, started_at VARCHAR(35),finished_at VARCHAR(35), UNIQUE KEY uq_ai_match_batch_syb (batch_id,syb_id), KEY idx_ai_match_batch_work (batch_id,status,leader_syb_id,position), KEY idx_ai_match_batch_identity (batch_id,identity_hash,position), CONSTRAINT fk_ai_match_item_batch FOREIGN KEY (batch_id) REFERENCES ai_match_batches(batch_id) ON DELETE CASCADE, CONSTRAINT chk_ai_match_item_status CHECK (status IN ('queued','running','succeeded','reused','manual','failed','stale','interrupted')), CONSTRAINT chk_ai_match_item_source CHECK (mapping_source IS NULL OR mapping_source IN ('manual','rule','ai')), CONSTRAINT chk_ai_match_item_confidence CHECK (confidence_bps IS NULL OR confidence_bps BETWEEN 0 AND 10000), CONSTRAINT chk_ai_match_item_called CHECK (model_called IN (0,1)) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, } for _, statement := range statements { if _, err := db.Exec(statement); err != nil { return err } } return nil } // migrateMySQLV21 给当前有效规格映射增加来源，并建立只追加的 AI 决策审计。 func migrateMySQLV21(db *sql.DB) error { columns := []struct{ name, ddl string }{ {"source", `ALTER TABLE spec_mappings ADD COLUMN source VARCHAR(16) NOT NULL DEFAULT 'manual' AFTER mapped_by`}, {"source_provider_id", `ALTER TABLE spec_mappings ADD COLUMN source_provider_id VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER source`}, {"source_model", `ALTER TABLE spec_mappings ADD COLUMN source_model VARCHAR(191) NULL AFTER source_provider_id`}, {"confidence_bps", `ALTER TABLE spec_mappings ADD COLUMN confidence_bps INT NULL AFTER source_model`}, {"source_reason", `ALTER TABLE spec_mappings ADD COLUMN source_reason VARCHAR(500) NULL AFTER confidence_bps`}, {"source_version", `ALTER TABLE spec_mappings ADD COLUMN source_version VARCHAR(64) NULL AFTER source_reason`}, {"context_version", `ALTER TABLE spec_mappings ADD COLUMN context_version CHAR(64) COLLATE utf8mb4_bin NULL AFTER source_version`}, } for _, column := range columns { exists, err := mysqlColumnExists(db, "spec_mappings", column.name) if err != nil { return err } if !exists { if _, err := db.Exec(column.ddl); err != nil { return err } } } for _, item := range []struct{ name, ddl string }{ {"chk_spec_mappings_source", `ALTER TABLE spec_mappings ADD CONSTRAINT chk_spec_mappings_source CHECK (source IN ('manual','rule','ai'))`}, {"chk_spec_mappings_confidence", `ALTER TABLE spec_mappings ADD CONSTRAINT chk_spec_mappings_confidence CHECK (confidence_bps IS NULL OR confidence_bps BETWEEN 0 AND 10000)`}, } { exists, err := mysqlConstraintExists(db, "spec_mappings", item.name) if err != nil { return err } if !exists { if _, err := db.Exec(item.ddl); err != nil { return err } } } _, err := db.Exec(`CREATE TABLE IF NOT EXISTS ai_spec_match_decisions ( id BIGINT AUTO_INCREMENT PRIMARY KEY, shopee_goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, spec_key VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, pdd_goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, context_version CHAR(64) COLLATE utf8mb4_bin NOT NULL, rules_version VARCHAR(32) NOT NULL, prompt_version VARCHAR(32) NOT NULL, provider_id VARCHAR(191) COLLATE utf8mb4_bin, provider_name VARCHAR(191), model VARCHAR(191), config_fingerprint CHAR(64) COLLATE utf8mb4_bin, candidates_json LONGTEXT NOT NULL, chosen_candidate_id VARCHAR(32) COLLATE utf8mb4_bin, chosen_option_key VARCHAR(191) COLLATE utf8mb4_bin, confidence_bps INT, outcome VARCHAR(32) NOT NULL, reason VARCHAR(500), conflict_dimensions_json LONGTEXT NOT NULL, missing_dimensions_json LONGTEXT NOT NULL, decided_by VARCHAR(191) COLLATE utf8mb4_bin, decided_at VARCHAR(35) NOT NULL, KEY idx_ai_match_identity (shopee_goods_id,spec_key,pdd_goods_id,decided_at DESC,id DESC), KEY idx_ai_match_outcome (outcome,decided_at DESC,id DESC), CONSTRAINT fk_ai_match_actor FOREIGN KEY (decided_by) REFERENCES users(user_id), CONSTRAINT chk_ai_match_confidence CHECK (confidence_bps IS NULL OR confidence_bps BETWEEN 0 AND 10000), CONSTRAINT chk_ai_match_outcome CHECK (outcome IN ('rule_saved','ai_saved','manual_exists','reused','rejected','failed','stale')) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`) return err } // migrateMySQLV20 保存 AI 服务商的非敏感配置和配置审计。 // API Key 只存在独立密钥文件中，不在这两张表里留列。 func migrateMySQLV20(db *sql.DB) error { statements := []string{ `CREATE TABLE IF NOT EXISTS ai_provider_configs ( provider_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, name VARCHAR(191) NOT NULL, base_url VARCHAR(2048) NOT NULL, model VARCHAR(191) NOT NULL, timeout_seconds INT NOT NULL, max_concurrency INT NOT NULL, confidence_threshold_bps INT NOT NULL, enabled TINYINT NOT NULL DEFAULT 0, active_slot TINYINT GENERATED ALWAYS AS (IF(enabled=1,1,NULL)) STORED, last_test_status VARCHAR(16) NOT NULL DEFAULT 'pending', last_test_message VARCHAR(500), last_tested_at VARCHAR(35), last_test_fingerprint CHAR(64) COLLATE utf8mb4_bin, created_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, updated_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL, UNIQUE KEY uq_ai_provider_name (name), UNIQUE KEY uq_ai_provider_active (active_slot), KEY idx_ai_provider_updated (updated_at DESC,provider_id), CONSTRAINT fk_ai_provider_created_by FOREIGN KEY (created_by_user_id) REFERENCES users(user_id), CONSTRAINT fk_ai_provider_updated_by FOREIGN KEY (updated_by_user_id) REFERENCES users(user_id), CONSTRAINT chk_ai_provider_enabled CHECK (enabled IN (0,1)), CONSTRAINT chk_ai_provider_timeout CHECK (timeout_seconds BETWEEN 1 AND 120), CONSTRAINT chk_ai_provider_concurrency CHECK (max_concurrency BETWEEN 1 AND 16), CONSTRAINT chk_ai_provider_confidence CHECK (confidence_threshold_bps BETWEEN 0 AND 10000), CONSTRAINT chk_ai_provider_test_status CHECK (last_test_status IN ('pending','succeeded','failed')) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS ai_provider_audits ( id BIGINT AUTO_INCREMENT PRIMARY KEY, provider_id VARCHAR(191) COLLATE utf8mb4_bin, action VARCHAR(32) NOT NULL, details_json LONGTEXT NOT NULL, actor_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, created_at VARCHAR(35) NOT NULL, KEY idx_ai_provider_audit (provider_id,created_at DESC,id DESC), CONSTRAINT fk_ai_audit_provider FOREIGN KEY (provider_id) REFERENCES ai_provider_configs(provider_id) ON DELETE SET NULL, CONSTRAINT fk_ai_audit_actor FOREIGN KEY (actor_user_id) REFERENCES users(user_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, } for _, statement := range statements { if _, err := db.Exec(statement); err != nil { return err } } return nil } // migrateMySQLV15 以低优先级补全历史 SYB 商品的店铺和图片。 func migrateMySQLV15(db *sql.DB) error { return backfillSybProductMetadata(db) } // migrateMySQLV16 为历史 SYB 数据补建格式明确的低优先级蝦皮 SKU。 func migrateMySQLV16(db *sql.DB) error { return backfillSybShopeeSKUs(db) } // migrateMySQLV17 为蝦皮商品增加可恢复的软删除标记。 func migrateMySQLV17(db *sql.DB) error { for _, column := range []struct{ name, ddl string }{ {"deleted_at", `ALTER TABLE shopee_products ADD COLUMN deleted_at VARCHAR(35) NULL AFTER pdd_goods_id`}, {"deleted_by_user_id", `ALTER TABLE shopee_products ADD COLUMN deleted_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER deleted_at`}, } { exists, err := mysqlColumnExists(db, "shopee_products", column.name) if err != nil { return err } if !exists { if _, err := db.Exec(column.ddl); err != nil { return fmt.Errorf("增加 shopee_products.%s 失败: %w", column.name, err) } } } if exists, err := mysqlIndexExists(db, "shopee_products", "idx_shopee_products_deleted"); err != nil { return err } else if !exists { if _, err := db.Exec(`ALTER TABLE shopee_products ADD INDEX idx_shopee_products_deleted (deleted_at,updated_at,goods_id)`); err != nil { return fmt.Errorf("增加蝦皮商品删除状态索引失败: %w", err) } } if exists, err := mysqlConstraintExists(db, "shopee_products", "fk_shopee_products_deleted_by"); err != nil { return err } else if !exists { if _, err := db.Exec(`ALTER TABLE shopee_products ADD CONSTRAINT fk_shopee_products_deleted_by FOREIGN KEY (deleted_by_user_id) REFERENCES users(user_id) ON DELETE SET NULL`); err != nil { return fmt.Errorf("增加蝦皮商品删除人外键失败: %w", err) } } return nil } // migrateMySQLV18 建立跨渠道店铺身份，保留旧 SYB 准入表作为回退依据。 func migrateMySQLV18(db *sql.DB) error { statements := []string{ `CREATE TABLE IF NOT EXISTS shops ( shop_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, display_name VARCHAR(500) NOT NULL, normalized_name VARCHAR(500) COLLATE utf8mb4_bin NOT NULL, enabled TINYINT NOT NULL DEFAULT 1, created_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL, UNIQUE KEY uq_shops_name (normalized_name), KEY idx_shops_enabled (enabled,normalized_name), CONSTRAINT fk_shops_user FOREIGN KEY (created_by_user_id) REFERENCES users(user_id), CONSTRAINT chk_shops_enabled CHECK (enabled IN (0,1)) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, `CREATE TABLE IF NOT EXISTS shop_channel_aliases ( alias_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, shop_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, channel VARCHAR(16) COLLATE utf8mb4_bin NOT NULL, alias_name VARCHAR(500) NOT NULL, normalized_alias VARCHAR(500) COLLATE utf8mb4_bin NOT NULL, enabled TINYINT NOT NULL DEFAULT 1, created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL, UNIQUE KEY uq_shop_channel_alias (channel,normalized_alias), KEY idx_shop_alias_shop (shop_id,channel,enabled), CONSTRAINT fk_shop_alias_shop FOREIGN KEY (shop_id) REFERENCES shops(shop_id) ON DELETE CASCADE, CONSTRAINT chk_shop_alias_channel CHECK (channel IN ('syb','shopee')), CONSTRAINT chk_shop_alias_enabled CHECK (enabled IN (0,1)) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`, } for _, statement := range statements { if _, err := db.Exec(statement); err != nil { return err } } for _, column := range []struct{ table, name, ddl string }{ {"syb_orders", "shop_id", `ALTER TABLE syb_orders ADD COLUMN shop_id VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER order_no`}, {"shopee_products", "shop_id", `ALTER TABLE shopee_products ADD COLUMN shop_id VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER goods_id`}, } { exists, err := mysqlColumnExists(db, column.table, column.name) if err != nil { return err } if !exists { if _, err := db.Exec(column.ddl); err != nil { return fmt.Errorf("增加 %s.%s 失败: %w", column.table, column.name, err) } } } for _, index := range []struct{ table, name, ddl string }{ {"syb_orders", "idx_syb_orders_shop_id", `ALTER TABLE syb_orders ADD INDEX idx_syb_orders_shop_id (shop_id)`}, {"shopee_products", "idx_shopee_products_shop_id", `ALTER TABLE shopee_products ADD INDEX idx_shopee_products_shop_id (shop_id)`}, } { exists, err := mysqlIndexExists(db, index.table, index.name) if err != nil { return err } if !exists { if _, err := db.Exec(index.ddl); err != nil { return err } } } for _, fk := range []struct{ table, name, ddl string }{ {"syb_orders", "fk_syb_orders_shop", `ALTER TABLE syb_orders ADD CONSTRAINT fk_syb_orders_shop FOREIGN KEY (shop_id) REFERENCES shops(shop_id) ON DELETE SET NULL`}, {"shopee_products", "fk_shopee_products_shop", `ALTER TABLE shopee_products ADD CONSTRAINT fk_shopee_products_shop FOREIGN KEY (shop_id) REFERENCES shops(shop_id) ON DELETE SET NULL`}, } { exists, err := mysqlConstraintExists(db, fk.table, fk.name) if err != nil { return err } if !exists { if _, err := db.Exec(fk.ddl); err != nil { return err } } } // 旧准入项一对一迁移成业务店铺；旧 enabled 只控制 SYB 渠道，不停用业务店铺。 if _, err := db.Exec(`INSERT INTO shops(shop_id,display_name,normalized_name,enabled,created_by_user_id,created_at,updated_at) SELECT shop_id,shop_name,normalized_name,1,created_by_user_id,created_at,updated_at FROM syb_allowed_shops ON DUPLICATE KEY UPDATE display_name=VALUES(display_name),updated_at=VALUES(updated_at)`); err != nil { return fmt.Errorf("迁移旧同步店铺失败: %w", err) } if _, err := db.Exec(`INSERT INTO shop_channel_aliases(alias_id,shop_id,channel,alias_name,normalized_alias,enabled,created_at,updated_at) SELECT CONCAT('legacy-syb-',shop_id),shop_id,'syb',shop_name,normalized_name,enabled,created_at,updated_at FROM syb_allowed_shops ON DUPLICATE KEY UPDATE shop_id=VALUES(shop_id),alias_name=VALUES(alias_name),enabled=VALUES(enabled),updated_at=VALUES(updated_at)`); err != nil { return fmt.Errorf("迁移 SYB 店铺别名失败: %w", err) } if _, err := db.Exec(`INSERT INTO shop_channel_aliases(alias_id,shop_id,channel,alias_name,normalized_alias,enabled,created_at,updated_at) SELECT CONCAT('legacy-shopee-',shop_id),shop_id,'shopee',shop_name,normalized_name,1,created_at,updated_at FROM syb_allowed_shops ON DUPLICATE KEY UPDATE shop_id=VALUES(shop_id),alias_name=VALUES(alias_name),updated_at=VALUES(updated_at)`); err != nil { return fmt.Errorf("迁移蝦皮店铺别名失败: %w", err) } if _, err := db.Exec(`UPDATE syb_orders so JOIN shop_channel_aliases a ON a.channel='syb' AND a.normalized_alias=TRIM(so.shop_name) COLLATE utf8mb4_bin SET so.shop_id=a.shop_id WHERE so.shop_id IS NULL`); err != nil { return fmt.Errorf("回填顺运宝业务店铺失败: %w", err) } if _, err := db.Exec(`UPDATE shopee_products sp JOIN shop_channel_aliases a ON a.channel='shopee' AND a.normalized_alias=TRIM(sp.shopee_shop_name) COLLATE utf8mb4_bin SET sp.shop_id=a.shop_id WHERE sp.shop_id IS NULL`); err != nil { return fmt.Errorf("回填蝦皮业务店铺失败: %w", err) } return nil } // migrateMySQLV19 把 v18 的渠道别名收敛成唯一店铺名称。 // shop_channel_aliases 暂时作为兼容表保留，其内容完全由 shops 派生，不再允许独立配置。 func migrateMySQLV19(db *sql.DB) error { tx, err := db.Begin() if err != nil { return err } defer tx.Rollback() // v18 同时存在全局启停和 SYB 启停。收敛时任一侧已停用都保持停用， // 避免升级后意外把原本禁止同步的店铺重新放入同步范围。 if _, err := tx.Exec(`UPDATE shops s JOIN ( SELECT shop_id,MIN(enabled) AS enabled FROM shop_channel_aliases WHERE channel='syb' GROUP BY shop_id ) a ON a.shop_id=s.shop_id SET s.enabled=IF(s.enabled=1 AND a.enabled=1,1,0)`); err != nil { return fmt.Errorf("收敛店铺启停状态失败: %w", err) } if _, err := tx.Exec(`DELETE FROM shop_channel_aliases`); err != nil { return fmt.Errorf("清理旧渠道店铺名称失败: %w", err) } for _, channel := range []string{"syb", "shopee"} { if _, err := tx.Exec(`INSERT INTO shop_channel_aliases (alias_id,shop_id,channel,alias_name,normalized_alias,enabled,created_at,updated_at) SELECT CONCAT('v19-',?,'-',SHA2(shop_id,256)),shop_id,?,display_name,normalized_name,enabled,created_at,updated_at FROM shops`, channel, channel); err != nil { return fmt.Errorf("生成 %s 店铺兼容数据失败: %w", channel, err) } } if _, err := tx.Exec(`UPDATE syb_orders SET shop_id=NULL`); err != nil { return fmt.Errorf("清理顺运宝旧店铺关联失败: %w", err) } if _, err := tx.Exec(`UPDATE syb_orders so JOIN shops s ON s.normalized_name=TRIM(so.shop_name) COLLATE utf8mb4_bin SET so.shop_id=s.shop_id`); err != nil { return fmt.Errorf("按唯一店铺名称重建顺运宝关联失败: %w", err) } if _, err := tx.Exec(`UPDATE shopee_products SET shop_id=NULL`); err != nil { return fmt.Errorf("清理蝦皮旧店铺关联失败: %w", err) } if _, err := tx.Exec(`UPDATE shopee_products sp JOIN shops s ON s.normalized_name=TRIM(sp.shopee_shop_name) COLLATE utf8mb4_bin SET sp.shop_id=s.shop_id`); err != nil { return fmt.Errorf("按唯一店铺名称重建蝦皮关联失败: %w", err) } return tx.Commit() } // migrateMySQLV14 增加顺运宝同步店铺准入表和审计统计。 func migrateMySQLV14(db *sql.DB) error { if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS syb_allowed_shops ( shop_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY, shop_name VARCHAR(500) NOT NULL, normalized_name VARCHAR(500) COLLATE utf8mb4_bin NOT NULL, enabled TINYINT NOT NULL DEFAULT 1, created_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, created_at VARCHAR(35) NOT NULL, updated_at VARCHAR(35) NOT NULL, UNIQUE KEY uq_syb_allowed_shops_name (normalized_name), KEY idx_syb_allowed_shops_enabled (enabled, normalized_name), CONSTRAINT fk_syb_allowed_shops_user FOREIGN KEY (created_by_user_id) REFERENCES users(user_id), CONSTRAINT chk_syb_allowed_shops_enabled CHECK (enabled IN (0,1)) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`); err != nil { return fmt.Errorf("建立顺运宝允许店铺表失败: %w", err) } columns := []struct{ name, ddl string }{ {"accepted_stock_count", `ALTER TABLE syb_sync_runs ADD COLUMN accepted_stock_count BIGINT NOT NULL DEFAULT 0 AFTER stock_count`}, {"shop_skipped_count", `ALTER TABLE syb_sync_runs ADD COLUMN shop_skipped_count BIGINT NOT NULL DEFAULT 0 AFTER accepted_stock_count`}, {"shop_filter_hash", `ALTER TABLE syb_sync_runs ADD COLUMN shop_filter_hash VARCHAR(64) COLLATE utf8mb4_bin NULL AFTER shop_skipped_count`}, } for _, column := range columns { exists, err := mysqlColumnExists(db, "syb_sync_runs", column.name) if err != nil { return err } if !exists { if _, err := db.Exec(column.ddl); err != nil { return fmt.Errorf("增加 syb_sync_runs.%s 失败: %w", column.name, err) } } } if _, err := db.Exec(`UPDATE syb_sync_runs SET accepted_stock_count=stock_count, shop_skipped_count=0 WHERE shop_filter_hash IS NULL`); err != nil { return fmt.Errorf("回填顺运宝同步店铺统计失败: %w", err) } return nil } // 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 { 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 { exists, err := mysqlColumnExists(db, "syb_orders", "shop_name") if err != nil { return err } if !exists { if _, err := db.Exec(`ALTER TABLE syb_orders ADD COLUMN shop_name VARCHAR(500) NULL AFTER order_no`); err != nil { return fmt.Errorf("增加 syb_orders.shop_name 失败: %w", err) } } if _, err := db.Exec(`UPDATE syb_orders SET shop_name=NULLIF(TRIM(JSON_UNQUOTE(JSON_EXTRACT( IF(JSON_VALID(syb_data),syb_data,'{}'),'$.stock.shopName'))),'') WHERE shop_name IS NULL`); err != nil { return fmt.Errorf("回填顺运宝店铺名失败: %w", err) } return nil } func checkMySQLV11Shape(db *sql.DB) error { if err := checkMySQLVarcharColumn(db, "syb_orders", "shop_name", 500, true, "utf8mb4_0900_ai_ci", ""); err != nil { return err } return checkMySQLNullDefault(db, "syb_orders", "shop_name") } // migrateMySQLV10 修复曾在 v8/v9 开发中间状态记录过版本的数据库。 // 两版迁移均逐项检查结构，v8 回填只处理空值，因此可安全重放； // 历史 v8/v9 版本记录保持不变，完整自检通过后再追加 v10。 func migrateMySQLV10(db *sql.DB) error { if err := migrateMySQLV8(db); err != nil { return fmt.Errorf("修复商品目录 SKU v8 结构失败: %w", err) } if err := migrateMySQLV9(db); err != nil { return fmt.Errorf("修复蝦皮商品 v9 结构失败: %w", err) } return nil } func checkMySQLV10Shape(db *sql.DB) error { if err := checkMySQLV8Shape(db); err != nil { return err } if err := checkMySQLV9Shape(db); err != nil { return err } return nil } func migrateMySQLV9(db *sql.DB) error { columns := []struct{ table, name, ddl string }{ {"shopee_products", "image_url", `ALTER TABLE shopee_products ADD COLUMN image_url VARCHAR(2048) NULL AFTER main_sku_code`}, {"shopee_products", "shopee_shop_name", `ALTER TABLE shopee_products ADD COLUMN shopee_shop_name VARCHAR(500) NULL AFTER image_url`}, {"shopee_products", "image_source", `ALTER TABLE shopee_products ADD COLUMN image_source VARCHAR(64) COLLATE utf8mb4_bin NULL AFTER shopee_shop_name`}, {"shopee_products", "image_observed_at", `ALTER TABLE shopee_products ADD COLUMN image_observed_at VARCHAR(35) NULL AFTER image_source`}, {"shopee_products", "image_is_manual", `ALTER TABLE shopee_products ADD COLUMN image_is_manual TINYINT NOT NULL DEFAULT 0 AFTER image_observed_at`}, {"shopee_products", "shop_name_source", `ALTER TABLE shopee_products ADD COLUMN shop_name_source VARCHAR(64) COLLATE utf8mb4_bin NULL AFTER image_is_manual`}, {"shopee_products", "shop_name_observed_at", `ALTER TABLE shopee_products ADD COLUMN shop_name_observed_at VARCHAR(35) NULL AFTER shop_name_source`}, {"shopee_products", "shop_name_is_manual", `ALTER TABLE shopee_products ADD COLUMN shop_name_is_manual TINYINT NOT NULL DEFAULT 0 AFTER shop_name_observed_at`}, {"catalog_import_runs", "shopee_fields_filled", `ALTER TABLE catalog_import_runs ADD COLUMN shopee_fields_filled INT UNSIGNED NOT NULL DEFAULT 0 AFTER shopee_updated`}, {"catalog_import_runs", "shopee_fields_same_source_updated", `ALTER TABLE catalog_import_runs ADD COLUMN shopee_fields_same_source_updated INT UNSIGNED NOT NULL DEFAULT 0 AFTER shopee_fields_filled`}, {"catalog_import_runs", "shopee_fields_manual_skipped", `ALTER TABLE catalog_import_runs ADD COLUMN shopee_fields_manual_skipped INT UNSIGNED NOT NULL DEFAULT 0 AFTER shopee_fields_same_source_updated`}, {"catalog_import_runs", "shopee_fields_stale_skipped", `ALTER TABLE catalog_import_runs ADD COLUMN shopee_fields_stale_skipped INT UNSIGNED NOT NULL DEFAULT 0 AFTER shopee_fields_manual_skipped`}, } for _, column := range columns { exists, err := mysqlColumnExists(db, column.table, column.name) if err != nil { return err } if !exists { if _, err := db.Exec(column.ddl); err != nil { return fmt.Errorf("增加 %s.%s 失败: %w", column.table, column.name, err) } } } for _, constraint := range []struct{ name, ddl string }{ {"chk_shopee_products_image_manual", `ALTER TABLE shopee_products ADD CONSTRAINT chk_shopee_products_image_manual CHECK (image_is_manual IN (0,1))`}, {"chk_shopee_products_shop_manual", `ALTER TABLE shopee_products ADD CONSTRAINT chk_shopee_products_shop_manual CHECK (shop_name_is_manual IN (0,1))`}, } { exists, err := mysqlConstraintExists(db, "shopee_products", constraint.name) if err != nil { return err } if !exists { if _, err := db.Exec(constraint.ddl); err != nil { return fmt.Errorf("增加约束 %s 失败: %w", constraint.name, err) } } } return nil } // migrateMySQLV8 保留 sku_id 作为内部主键，并增加可空的真实蝦皮 SKU ID。 // 每个 DDL 都先检查形状，支持 MySQL 隐式提交后的中断重放。 func migrateMySQLV8(db *sql.DB) error { columns := []struct{ table, name, ddl string }{ {"shopee_skus", "shopee_sku_id", `ALTER TABLE shopee_skus ADD COLUMN shopee_sku_id VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER sku_id`}, {"shopee_skus", "spec_key", `ALTER TABLE shopee_skus ADD COLUMN spec_key VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER spec_raw`}, {"shopee_skus", "source", `ALTER TABLE shopee_skus ADD COLUMN source VARCHAR(64) COLLATE utf8mb4_bin NULL AFTER is_manual`}, {"shopee_skus", "field_sources", `ALTER TABLE shopee_skus ADD COLUMN field_sources JSON NULL AFTER source`}, {"shopee_skus", "field_observed_at", `ALTER TABLE shopee_skus ADD COLUMN field_observed_at JSON NULL AFTER field_sources`}, {"catalog_import_runs", "update_policy", `ALTER TABLE catalog_import_runs ADD COLUMN update_policy VARCHAR(32) COLLATE utf8mb4_bin NOT NULL DEFAULT 'fill_missing' AFTER observed_at`}, {"catalog_import_runs", "sku_filled", `ALTER TABLE catalog_import_runs ADD COLUMN sku_filled INT UNSIGNED NOT NULL DEFAULT 0 AFTER sku_updated`}, {"catalog_import_runs", "sku_same_source_updated", `ALTER TABLE catalog_import_runs ADD COLUMN sku_same_source_updated INT UNSIGNED NOT NULL DEFAULT 0 AFTER sku_filled`}, {"catalog_import_runs", "sku_skipped", `ALTER TABLE catalog_import_runs ADD COLUMN sku_skipped INT UNSIGNED NOT NULL DEFAULT 0 AFTER sku_same_source_updated`}, {"catalog_import_runs", "sku_manual_skipped", `ALTER TABLE catalog_import_runs ADD COLUMN sku_manual_skipped INT UNSIGNED NOT NULL DEFAULT 0 AFTER sku_skipped`}, {"catalog_import_runs", "sku_stale_skipped", `ALTER TABLE catalog_import_runs ADD COLUMN sku_stale_skipped INT UNSIGNED NOT NULL DEFAULT 0 AFTER sku_manual_skipped`}, } for _, column := range columns { exists, err := mysqlColumnExists(db, column.table, column.name) if err != nil { return err } if !exists { if _, err := db.Exec(column.ddl); err != nil { return fmt.Errorf("增加 %s.%s 失败: %w", column.table, column.name, err) } } } if _, err := db.Exec(`UPDATE shopee_skus SET shopee_sku_id=sku_id WHERE shopee_sku_id IS NULL AND sku_id NOT LIKE 'catalog:%'`); err != nil { return fmt.Errorf("回填真实蝦皮 SKU ID 失败: %w", err) } if _, err := db.Exec(`UPDATE shopee_skus SET source='report' WHERE source IS NULL OR source=''`); err != nil { return fmt.Errorf("回填 SKU 来源失败: %w", err) } if _, err := db.Exec(`UPDATE shopee_skus SET field_sources=JSON_OBJECT('color',IF(color IS NULL OR color='','',COALESCE(source,'report')),'size',IF(size IS NULL OR size='','',COALESCE(source,'report')),'advice',IF(advice IS NULL OR advice='','',COALESCE(source,'report')),'sku_code',IF(sku_code IS NULL OR sku_code='','',COALESCE(source,'report'))) WHERE field_sources IS NULL`); err != nil { return fmt.Errorf("回填 SKU 字段来源失败: %w", err) } if _, err := db.Exec(`UPDATE shopee_skus SET field_observed_at=JSON_OBJECT('color',IF(color IS NULL OR color='','',COALESCE(source_observed_at,updated_at)),'size',IF(size IS NULL OR size='','',COALESCE(source_observed_at,updated_at)),'advice',IF(advice IS NULL OR advice='','',COALESCE(source_observed_at,updated_at)),'sku_code',IF(sku_code IS NULL OR sku_code='','',COALESCE(source_observed_at,updated_at))) WHERE field_observed_at IS NULL`); err != nil { return fmt.Errorf("回填 SKU 字段观测时间失败: %w", err) } rows, err := db.Query(`SELECT sku_id,spec_raw FROM shopee_skus WHERE spec_key IS NULL OR spec_key=''`) if err != nil { return fmt.Errorf("读取待回填规格键失败: %w", err) } type pendingKey struct{ id, raw string } var pending []pendingKey for rows.Next() { var row pendingKey if err := rows.Scan(&row.id, &row.raw); err != nil { rows.Close() return err } pending = append(pending, row) } if err := rows.Close(); err != nil { return err } for _, row := range pending { key, keyErr := spec.SpecKey(row.raw) if keyErr != nil { return fmt.Errorf("SKU 内部记录 %s 无法生成规格键: %w", row.id, keyErr) } if len([]rune(key)) > 191 { return fmt.Errorf("SKU 内部记录 %s 的规格键超过 191 个字符", row.id) } if _, err := db.Exec(`UPDATE shopee_skus SET spec_key=? WHERE sku_id=?`, key, row.id); err != nil { return err } } var duplicate string err = db.QueryRow(`SELECT CONCAT(goods_id,' / ',spec_key) FROM shopee_skus GROUP BY goods_id,spec_key HAVING COUNT(*)>1 LIMIT 1`).Scan(&duplicate) if err != nil && err != sql.ErrNoRows { return err } if err == nil { return fmt.Errorf("存在同商品重复规格 %s，请人工确认真实 SKU 后再迁移", duplicate) } for _, index := range []struct{ name, ddl string }{ {"uq_shopee_skus_external", `ALTER TABLE shopee_skus ADD UNIQUE KEY uq_shopee_skus_external (shopee_sku_id)`}, {"uq_shopee_skus_spec", `ALTER TABLE shopee_skus ADD UNIQUE KEY uq_shopee_skus_spec (goods_id,spec_key)`}, } { exists, err := mysqlIndexExists(db, "shopee_skus", index.name) if err != nil { return err } if !exists { if _, err := db.Exec(index.ddl); err != nil { return fmt.Errorf("增加索引 %s 失败: %w", index.name, err) } } } return nil } const mysqlSchemaV7CatalogRuns = `CREATE TABLE IF NOT EXISTS catalog_import_runs ( source VARCHAR(64) COLLATE utf8mb4_bin NOT NULL, batch_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, request_hash VARCHAR(64) COLLATE utf8mb4_bin NOT NULL, status VARCHAR(16) COLLATE utf8mb4_bin NOT NULL DEFAULT 'processing', request_count INT UNSIGNED NOT NULL DEFAULT 1, conflict_count INT UNSIGNED NOT NULL DEFAULT 0, observed_at VARCHAR(35) NULL, last_request_at VARCHAR(35) NOT NULL, last_conflict_at VARCHAR(35) NULL, shopee_created INT UNSIGNED NOT NULL DEFAULT 0, shopee_updated INT UNSIGNED NOT NULL DEFAULT 0, sku_created INT UNSIGNED NOT NULL DEFAULT 0, sku_updated INT UNSIGNED NOT NULL DEFAULT 0, pdd_created INT UNSIGNED NOT NULL DEFAULT 0, pdd_updated INT UNSIGNED NOT NULL DEFAULT 0, association_created INT UNSIGNED NOT NULL DEFAULT 0, association_unchanged INT UNSIGNED NOT NULL DEFAULT 0, failure_count INT UNSIGNED NOT NULL DEFAULT 0, error_summary VARCHAR(1000) NULL, response_body LONGTEXT NULL, created_at VARCHAR(35) NOT NULL, finished_at VARCHAR(35) NULL, PRIMARY KEY (source, batch_id), KEY idx_catalog_runs_list (created_at DESC, source, batch_id), KEY idx_catalog_runs_status (status, created_at DESC), CONSTRAINT chk_catalog_runs_status CHECK (status IN ('processing','succeeded','failed')) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci` // migrateMySQLV7 建立第三方商品目录批次基础。所有 DDL 均逐项检查，支持隐式提交后重放。 func migrateMySQLV7(db *sql.DB) error { columns := []struct { table, name, ddl string }{ {"shopee_products", "source_observed_at", `ALTER TABLE shopee_products ADD COLUMN source_observed_at VARCHAR(35) NULL AFTER source`}, {"shopee_skus", "source_observed_at", `ALTER TABLE shopee_skus ADD COLUMN source_observed_at VARCHAR(35) NULL AFTER is_manual`}, {"pdd_products", "source", `ALTER TABLE pdd_products ADD COLUMN source VARCHAR(16) COLLATE utf8mb4_bin NULL AFTER collect_status`}, {"pdd_products", "source_observed_at", `ALTER TABLE pdd_products ADD COLUMN source_observed_at VARCHAR(35) NULL AFTER source`}, } for _, column := range columns { exists, err := mysqlColumnExists(db, column.table, column.name) if err != nil { return err } if !exists { if _, err := db.Exec(column.ddl); err != nil { return fmt.Errorf("增加 %s.%s 失败: %w", column.table, column.name, err) } } } var clause string err := db.QueryRow(`SELECT cc.check_clause FROM information_schema.table_constraints tc JOIN information_schema.check_constraints cc ON cc.constraint_schema=tc.constraint_schema AND cc.constraint_name=tc.constraint_name WHERE tc.constraint_schema=DATABASE() AND tc.table_name='shopee_products' AND tc.constraint_name='chk_shopee_products_source' AND tc.constraint_type='CHECK'`).Scan(&clause) if err != nil && err != sql.ErrNoRows { return fmt.Errorf("读取蝦皮来源约束失败: %w", err) } normalized := strings.NewReplacer("`", "", " ", "", "(", "", ")", "", "_utf8mb4", "", `\`, "").Replace(strings.ToLower(clause)) if !strings.Contains(normalized, "'api'") { if err == nil { if _, dropErr := db.Exec(`ALTER TABLE shopee_products DROP CHECK chk_shopee_products_source`); dropErr != nil { return fmt.Errorf("移除旧蝦皮来源约束失败: %w", dropErr) } } if _, addErr := db.Exec(`ALTER TABLE shopee_products ADD CONSTRAINT chk_shopee_products_source CHECK (source IN ('report','syb','api'))`); addErr != nil { return fmt.Errorf("扩展蝦皮来源约束失败: %w", addErr) } } if _, err := db.Exec(mysqlSchemaV7CatalogRuns); err != nil { return fmt.Errorf("建立商品目录导入记录表失败: %w", err) } return nil } // migrateMySQLV6 给任务补充创建人。存量任务保持 NULL，明确标记为历史任务； // 新代码创建任务时写入当前网页登录用户。DDL 逐项检查，支持中断后重放。 func migrateMySQLV6(db *sql.DB) error { exists, err := mysqlColumnExists(db, "tasks", "created_by_user_id") if err != nil { return err } if !exists { if _, err := db.Exec(`ALTER TABLE tasks ADD COLUMN created_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER live_confirmed_at`); err != nil { return fmt.Errorf("增加 tasks.created_by_user_id 失败: %w", err) } } constraintExists, err := mysqlConstraintExists(db, "tasks", "fk_tasks_created_by") if err != nil { return err } if !constraintExists { if _, err := db.Exec(`ALTER TABLE tasks ADD CONSTRAINT fk_tasks_created_by FOREIGN KEY (created_by_user_id) REFERENCES users(user_id)`); err != nil { return fmt.Errorf("增加任务创建人外键失败: %w", err) } } indexExists, err := mysqlIndexExists(db, "tasks", "idx_tasks_creator_list") if err != nil { return err } if !indexExists { if _, err := db.Exec(`ALTER TABLE tasks ADD INDEX idx_tasks_creator_list (created_by_user_id, updated_at DESC, task_id DESC)`); err != nil { return fmt.Errorf("增加任务创建人列表索引失败: %w", err) } } return nil } // migrateMySQLV5 给任务增加不可变执行模式和真实下单创建审计。 // 每条 DDL 都先检查存在性，MySQL 在任意一步隐式提交后都可以安全重放。 func migrateMySQLV5(db *sql.DB) error { columns := []struct { name string ddl string }{ {"execution_mode", `ALTER TABLE tasks ADD COLUMN execution_mode VARCHAR(16) COLLATE utf8mb4_bin NOT NULL DEFAULT 'dry_run' AFTER status`}, {"live_confirmed_by", `ALTER TABLE tasks ADD COLUMN live_confirmed_by VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER finished_at`}, {"live_confirmed_at", `ALTER TABLE tasks ADD COLUMN live_confirmed_at VARCHAR(35) NULL AFTER live_confirmed_by`}, } for _, column := range columns { exists, err := mysqlColumnExists(db, "tasks", column.name) if err != nil { return err } if !exists { if _, err := db.Exec(column.ddl); err != nil { return fmt.Errorf("增加 tasks.%s 失败: %w", column.name, err) } } } if _, err := db.Exec(`UPDATE tasks SET execution_mode='dry_run' WHERE execution_mode IS NULL OR execution_mode=''`); err != nil { return fmt.Errorf("回填任务执行模式失败: %w", err) } constraints := []struct { name string ddl string }{ {"chk_tasks_execution_mode", `ALTER TABLE tasks ADD CONSTRAINT chk_tasks_execution_mode CHECK (execution_mode IN ('dry_run','live'))`}, {"chk_tasks_live_confirmation", `ALTER TABLE tasks ADD CONSTRAINT chk_tasks_live_confirmation CHECK ((execution_mode='dry_run' AND live_confirmed_by IS NULL AND live_confirmed_at IS NULL) OR (execution_mode='live' AND live_confirmed_by IS NOT NULL AND live_confirmed_by<>'' AND live_confirmed_at IS NOT NULL AND live_confirmed_at<>''))`}, } for _, constraint := range constraints { exists, err := mysqlConstraintExists(db, "tasks", constraint.name) if err != nil { return err } if !exists { if _, err := db.Exec(constraint.ddl); err != nil { return fmt.Errorf("增加 tasks.%s 失败: %w", constraint.name, err) } } } indexExists, err := mysqlIndexExists(db, "tasks", "idx_tasks_claim_mode") if err != nil { return err } if !indexExists { if _, err := db.Exec(`ALTER TABLE tasks ADD INDEX idx_tasks_claim_mode (assigned_client, status, execution_mode, priority DESC, created_at)`); err != nil { return fmt.Errorf("增加任务模式领取索引失败: %w", err) } } return nil } func checkMySQLSchemaV3(db *sql.DB) error { tables := []string{"shopee_products", "shopee_skus", "pdd_products", "syb_orders", "spec_mappings", "tasks", "clients", "idempotency_keys", "task_claims", "syb_session", "syb_sync_state", "users", "web_sessions", "client_user_assignments", "syb_sync_runs", "admin_initialization_lock"} if err := checkMySQLSchema(db, tables); err != nil { return err } return checkMySQLV3Shape(db) } // migrateMySQLV3 把采购规格身份从蝦皮 SKU 改为顺运宝商品规格原文。 // 每一步都可重放：DDL 已提交但版本尚未记录时，再启动仍会收敛。 func migrateMySQLV3(db *sql.DB) error { if err := ensureMySQLV3Columns(db); err != nil { return err } if _, err := db.Exec(mysqlSchemaV3SpecMappings); err != nil { return fmt.Errorf("建立 spec_mappings 失败: %w", err) } if err := backfillSybSpecKeys(db); err != nil { return err } if err := ensureSybShopeeSkeletons(db); err != nil { return err } return convertLegacySKUMappings(db) } func ensureMySQLV3Columns(db *sql.DB) error { exists, err := mysqlColumnExists(db, "syb_orders", "spec_key") if err != nil { return err } if !exists { if _, err := db.Exec(`ALTER TABLE syb_orders ADD COLUMN spec_key VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER product_spec`); err != nil { return fmt.Errorf("增加 syb_orders.spec_key 失败: %w", err) } } else if _, err := db.Exec(`ALTER TABLE syb_orders MODIFY COLUMN spec_key VARCHAR(191) COLLATE utf8mb4_bin NULL`); err != nil { return fmt.Errorf("校正 syb_orders.spec_key 结构失败: %w", err) } exists, err = mysqlColumnExists(db, "shopee_products", "source") if err != nil { return err } if !exists { if _, err := db.Exec(`ALTER TABLE shopee_products ADD COLUMN source VARCHAR(16) COLLATE utf8mb4_bin NOT NULL DEFAULT 'report' AFTER main_sku_code`); err != nil { return fmt.Errorf("增加 shopee_products.source 失败: %w", err) } } else { // 兼容“列已加但还是可空”的 DDL 中断点。先回填旧行， // 再收紧 NOT NULL，否则带 NULL 的存量数据会让 ALTER 失败。 if _, err := db.Exec(`UPDATE shopee_products SET source='report' WHERE source IS NULL OR source=''`); err != nil { return fmt.Errorf("回填 shopee_products.source 失败: %w", err) } if _, err := db.Exec(`ALTER TABLE shopee_products MODIFY COLUMN source VARCHAR(16) COLLATE utf8mb4_bin NOT NULL DEFAULT 'report'`); err != nil { return fmt.Errorf("校正 shopee_products.source 结构失败: %w", err) } } constraintExists, err := mysqlConstraintExists(db, "shopee_products", "chk_shopee_products_source") if err != nil { return err } if !constraintExists { if _, err := db.Exec(`ALTER TABLE shopee_products ADD CONSTRAINT chk_shopee_products_source CHECK (source IN ('report','syb'))`); err != nil { return fmt.Errorf("增加 shopee_products.source CHECK 失败: %w", err) } } return nil } func mysqlColumnExists(db *sql.DB, table, column string) (bool, error) { var count int if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name=? AND column_name=?`, table, column).Scan(&count); err != nil { return false, fmt.Errorf("检查 MySQL 数据表 %s.%s 失败: %w", table, column, err) } return count == 1, nil } func mysqlConstraintExists(db *sql.DB, table, constraint string) (bool, error) { var count int if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.table_constraints WHERE constraint_schema=DATABASE() AND table_name=? AND constraint_name=?`, table, constraint).Scan(&count); err != nil { return false, fmt.Errorf("检查 MySQL 约束 %s.%s 失败: %w", table, constraint, err) } return count == 1, nil } func mysqlIndexExists(db *sql.DB, table, index string) (bool, error) { var count int if err := db.QueryRow(`SELECT COUNT(DISTINCT index_name) FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name=? AND index_name=?`, table, index).Scan(&count); err != nil { return false, fmt.Errorf("检查 MySQL 索引 %s.%s 失败: %w", table, index, err) } return count == 1, nil } func backfillSybSpecKeys(db *sql.DB) error { const batchSize = 500 cursor := "" for { rows, err := db.Query(`SELECT syb_id, product_spec FROM syb_orders WHERE syb_id > ? AND spec_key IS NULL AND product_spec IS NOT NULL AND TRIM(product_spec) <> '' ORDER BY syb_id LIMIT ?`, cursor, batchSize) if err != nil { return fmt.Errorf("读取待回填顺运宝规格失败: %w", err) } type item struct{ id, raw, key string } items := make([]item, 0, batchSize) for rows.Next() { var value item if err := rows.Scan(&value.id, &value.raw); err != nil { rows.Close() return fmt.Errorf("读取待回填顺运宝规格失败: %w", err) } value.key, err = spec.SpecKey(value.raw) if err != nil { rows.Close() return fmt.Errorf("顺运宝明细 %s 的规格不能生成身份键: %w", value.id, err) } items = append(items, value) } if err := rows.Close(); err != nil { return fmt.Errorf("关闭顺运宝规格回填结果失败: %w", err) } if len(items) == 0 { return nil } tx, err := db.Begin() if err != nil { return fmt.Errorf("开始顺运宝规格回填事务失败: %w", err) } for _, value := range items { if _, err := tx.Exec(`UPDATE syb_orders SET spec_key=? WHERE syb_id=? AND spec_key IS NULL`, value.key, value.id); err != nil { tx.Rollback() return fmt.Errorf("回填顺运宝明细 %s 的规格键失败: %w", value.id, err) } } if err := tx.Commit(); err != nil { return fmt.Errorf("提交顺运宝规格回填失败: %w", err) } cursor = items[len(items)-1].id } } func ensureSybShopeeSkeletons(db *sql.DB) error { now := model.NowISO() _, err := db.Exec(`INSERT INTO shopee_products (goods_id, title, source, created_at, updated_at) SELECT shopee_goods_id, MAX(COALESCE(title,'')), 'syb', ?, ? FROM syb_orders WHERE shopee_goods_id IS NOT NULL AND shopee_goods_id <> '' GROUP BY shopee_goods_id ON DUPLICATE KEY UPDATE goods_id=VALUES(goods_id)`, now, now) if err != nil { return fmt.Errorf("按顺运宝数据补建蝦皮商品骨架失败: %w", err) } return nil } func convertLegacySKUMappings(db *sql.DB) error { exists, err := mysqlTableExists(db, "sku_mappings") if err != nil || !exists { return err } if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS sku_mappings_v3_backup LIKE sku_mappings`); err != nil { return fmt.Errorf("建立旧规格映射备份表失败: %w", err) } if _, err := db.Exec(`INSERT IGNORE INTO sku_mappings_v3_backup SELECT * FROM sku_mappings`); err != nil { return fmt.Errorf("备份旧规格映射失败: %w", err) } rows, err := db.Query(`SELECT m.shopee_sku_id, m.pdd_goods_id, m.pdd_option_key, m.pdd_options, m.mapped_at, m.mapped_by, sk.goods_id, sk.spec_raw FROM sku_mappings m LEFT JOIN shopee_skus sk ON sk.sku_id=m.shopee_sku_id`) if err != nil { return fmt.Errorf("读取旧规格映射失败: %w", err) } type legacy struct { skuID, pddID, optionKey, options, mappedAt string mappedBy, goodsID, raw sql.NullString } var records []legacy for rows.Next() { var value legacy if err := rows.Scan(&value.skuID, &value.pddID, &value.optionKey, &value.options, &value.mappedAt, &value.mappedBy, &value.goodsID, &value.raw); err != nil { rows.Close() return fmt.Errorf("读取旧规格映射失败: %w", err) } records = append(records, value) } if err := rows.Close(); err != nil { return fmt.Errorf("关闭旧规格映射结果失败: %w", err) } matched, noCurrentMatch, unable := 0, 0, 0 for _, value := range records { if !value.goodsID.Valid || !value.raw.Valid { unable++ log.Printf("mysql_v3_mapping_unconvertible shopee_sku_id=%s pdd_goods_id=%s", value.skuID, value.pddID) continue } key, err := spec.SpecKey(value.raw.String) if err != nil { unable++ log.Printf("mysql_v3_mapping_unconvertible shopee_sku_id=%s pdd_goods_id=%s reason=invalid_spec_key", value.skuID, value.pddID) continue } if err := UpsertSpecMapping(db, model.SpecMapping{ ShopeeGoodsID: value.goodsID.String, SpecKey: key, SpecRaw: value.raw.String, PddGoodsID: value.pddID, PddOptionKey: value.optionKey, PddOptions: value.options, MappedAt: value.mappedAt, MappedBy: value.mappedBy.String, }); err != nil { return fmt.Errorf("转换旧规格映射 %s 失败: %w", value.skuID, err) } var hit int if err := db.QueryRow(`SELECT EXISTS(SELECT 1 FROM syb_orders WHERE shopee_goods_id=? AND spec_key=?)`, value.goodsID.String, key).Scan(&hit); err != nil { return fmt.Errorf("验证旧规格映射 %s 命中失败: %w", value.skuID, err) } if hit == 1 { matched++ } else { noCurrentMatch++ log.Printf("mysql_v3_mapping_no_current_hit shopee_sku_id=%s pdd_goods_id=%s", value.skuID, value.pddID) } } if matched+noCurrentMatch+unable != len(records) { return fmt.Errorf("旧规格映射迁移计数不守恒: 总数=%d 命中=%d 无命中=%d 无法转换=%d", len(records), matched, noCurrentMatch, unable) } log.Printf("mysql_v3_mapping_summary total=%d matched=%d no_current_hit=%d unconvertible=%d backup_table=sku_mappings_v3_backup", len(records), matched, noCurrentMatch, unable) if _, err := db.Exec(`DROP TABLE sku_mappings`); err != nil { return fmt.Errorf("删除已备份的旧规格映射表失败: %w", err) } return nil } func mysqlTableExists(db *sql.DB, table string) (bool, error) { var count int if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name=? AND table_type='BASE TABLE'`, table).Scan(&count); err != nil { return false, fmt.Errorf("检查 MySQL 数据表 %s 失败: %w", table, err) } return count == 1, nil } // CheckMySQLSchema 确认所有业务表、关键追加列和采购身份结构完整。 func CheckMySQLSchema(db *sql.DB) error { mysqlRequiredTables := []string{ "shopee_products", "shopee_skus", "pdd_products", "syb_orders", "spec_mappings", "tasks", "clients", "idempotency_keys", "task_claims", "syb_session", "syb_sync_state", "users", "web_sessions", "client_user_assignments", "syb_sync_runs", "admin_initialization_lock", "spec_mapping_decisions", "catalog_import_runs", "task_syb_sources", "task_sequences", "syb_allowed_shops", "ai_provider_configs", "ai_provider_audits", "ai_spec_match_decisions", "ai_match_batches", "ai_match_batch_items", } if err := checkMySQLSchema(db, mysqlRequiredTables); err != nil { return err } if err := checkMySQLV3Shape(db); err != nil { return err } if err := checkMySQLV4Shape(db); err != nil { return err } if err := checkMySQLV5Shape(db); err != nil { return err } if err := checkMySQLV6Shape(db); err != nil { return err } if err := checkMySQLV7Shape(db); err != nil { return err } if err := checkMySQLV8Shape(db); err != nil { return err } if err := checkMySQLV9Shape(db); err != nil { return err } if err := checkMySQLV10Shape(db); err != nil { return err } if err := checkMySQLV11Shape(db); err != nil { return err } if err := checkMySQLV12Shape(db); err != nil { return err } if err := checkMySQLV13Shape(db); err != nil { return err } if err := checkMySQLV14Shape(db); err != nil { return err } if err := checkMySQLV15Shape(db); err != nil { return err } if err := checkMySQLV16Shape(db); err != nil { return err } if err := checkMySQLV17Shape(db); err != nil { return err } if err := checkMySQLV18Shape(db); err != nil { return err } if err := checkMySQLV19Shape(db); err != nil { return err } if err := checkMySQLV20Shape(db); err != nil { return err } if err := checkMySQLV21Shape(db); err != nil { return err } if err := checkMySQLV22Shape(db); err != nil { return err } if err := checkMySQLV23Shape(db); err != nil { return err } return checkMySQLV24Shape(db) } func checkMySQLV24Shape(db *sql.DB) error { for _, name := range []string{"deleted_at", "deleted_by_user_id"} { exists, err := mysqlColumnExists(db, "syb_inner_code_records", name) if err != nil || !exists { return fmt.Errorf("档口入库码软删除字段 %s 缺失: %v", name, err) } } for _, item := range []struct{ kind, name string }{ {"index", "idx_inner_code_deleted"}, {"constraint", "fk_inner_code_deleted_by"}, } { var exists bool var err error if item.kind == "index" { exists, err = mysqlIndexExists(db, "syb_inner_code_records", item.name) } else { exists, err = mysqlConstraintExists(db, "syb_inner_code_records", item.name) } if err != nil || !exists { return fmt.Errorf("档口入库码软删除%s %s 缺失: %v", item.kind, item.name, err) } } return nil } func checkMySQLV23Shape(db *sql.DB) error { if err := checkMySQLSchema(db, []string{"syb_inner_code_records"}); err != nil { return err } for _, name := range []string{"business_date", "order_number", "stall", "spec_key", "inner_code", "source_duplicate_count", "stock_id", "detail_id", "status", "created_by_user_id", "applied_by_user_id"} { exists, err := mysqlColumnExists(db, "syb_inner_code_records", name) if err != nil || !exists { return fmt.Errorf("档口入库码字段 %s 缺失: %v", name, err) } } for _, item := range []struct{ kind, name string }{ {"index", "uq_inner_code_business"}, {"index", "uq_inner_code_date_code"}, {"index", "idx_inner_code_list"}, {"constraint", "chk_inner_code_status"}, {"constraint", "fk_inner_code_created_by"}, } { var exists bool var err error if item.kind == "index" { exists, err = mysqlIndexExists(db, "syb_inner_code_records", item.name) } else { exists, err = mysqlConstraintExists(db, "syb_inner_code_records", item.name) } if err != nil || !exists { return fmt.Errorf("档口入库码%s %s 缺失: %v", item.kind, item.name, err) } } return nil } func checkMySQLV22Shape(db *sql.DB) error { if err := checkMySQLSchema(db, []string{"ai_match_batches", "ai_match_batch_items"}); err != nil { return err } for _, item := range []struct{ kind, table, name string }{ {"constraint", "ai_match_batches", "chk_ai_match_batch_status"}, {"constraint", "ai_match_batches", "chk_ai_match_batch_counts"}, {"constraint", "ai_match_batches", "fk_ai_match_batch_user"}, {"constraint", "ai_match_batch_items", "chk_ai_match_item_status"}, {"constraint", "ai_match_batch_items", "fk_ai_match_item_batch"}, {"index", "ai_match_batch_items", "uq_ai_match_batch_syb"}, {"index", "ai_match_batch_items", "idx_ai_match_batch_identity"}, } { var exists bool var err error if item.kind == "index" { exists, err = mysqlIndexExists(db, item.table, item.name) } else { exists, err = mysqlConstraintExists(db, item.table, item.name) } if err != nil || !exists { return fmt.Errorf("AI 匹配批次%s %s.%s 缺失: %v", item.kind, item.table, item.name, err) } } return nil } func checkMySQLV21Shape(db *sql.DB) error { if err := checkMySQLSchema(db, []string{"spec_mappings", "ai_spec_match_decisions"}); err != nil { return err } for _, name := range []string{"source", "source_provider_id", "source_model", "confidence_bps", "source_reason", "source_version", "context_version"} { exists, err := mysqlColumnExists(db, "spec_mappings", name) if err != nil || !exists { return fmt.Errorf("规格映射来源字段 %s 缺失: %v", name, err) } } for _, item := range []struct{ kind, table, name string }{ {"constraint", "spec_mappings", "chk_spec_mappings_source"}, {"constraint", "spec_mappings", "chk_spec_mappings_confidence"}, {"constraint", "ai_spec_match_decisions", "chk_ai_match_confidence"}, {"constraint", "ai_spec_match_decisions", "chk_ai_match_outcome"}, {"constraint", "ai_spec_match_decisions", "fk_ai_match_actor"}, {"index", "ai_spec_match_decisions", "idx_ai_match_identity"}, } { var exists bool var err error if item.kind == "index" { exists, err = mysqlIndexExists(db, item.table, item.name) } else { exists, err = mysqlConstraintExists(db, item.table, item.name) } if err != nil || !exists { return fmt.Errorf("AI 规格审计%s %s.%s 缺失: %v", item.kind, item.table, item.name, err) } } return nil } func checkMySQLV20Shape(db *sql.DB) error { if err := checkMySQLSchema(db, []string{"ai_provider_configs", "ai_provider_audits"}); err != nil { return err } for _, name := range []string{"provider_id", "base_url", "model", "timeout_seconds", "max_concurrency", "confidence_threshold_bps", "enabled", "active_slot", "last_test_status", "last_test_fingerprint"} { exists, err := mysqlColumnExists(db, "ai_provider_configs", name) if err != nil || !exists { return fmt.Errorf("AI 服务商配置字段 %s 缺失: %v", name, err) } } for _, item := range []struct{ kind, table, name string }{ {"index", "ai_provider_configs", "uq_ai_provider_name"}, {"index", "ai_provider_configs", "uq_ai_provider_active"}, {"constraint", "ai_provider_configs", "chk_ai_provider_test_status"}, {"constraint", "ai_provider_configs", "fk_ai_provider_created_by"}, {"constraint", "ai_provider_audits", "fk_ai_audit_actor"}, } { var exists bool var err error if item.kind == "index" { exists, err = mysqlIndexExists(db, item.table, item.name) } else { exists, err = mysqlConstraintExists(db, item.table, item.name) } if err != nil || !exists { return fmt.Errorf("AI 配置%s %s.%s 缺失: %v", item.kind, item.table, item.name, err) } } return nil } func checkMySQLV15Shape(db *sql.DB) error { rows, err := db.Query(`SELECT image_source,image_observed_at,image_is_manual, shop_name_source,shop_name_observed_at,shop_name_is_manual FROM shopee_products LIMIT 0`) if err != nil { return fmt.Errorf("蝦皮商品店铺与图片来源字段缺失: %w", err) } return rows.Close() } func checkMySQLV16Shape(db *sql.DB) error { rows, err := db.Query(`SELECT spec_key,color,size,advice,parse_ok,source,field_sources,field_observed_at FROM shopee_skus LIMIT 0`) if err != nil { return fmt.Errorf("蝦皮 SKU 解析与来源字段缺失: %w", err) } return rows.Close() } func checkMySQLV17Shape(db *sql.DB) error { if err := checkMySQLVarcharColumn(db, "shopee_products", "deleted_at", 35, true, "utf8mb4_0900_ai_ci", ""); err != nil { return err } if err := checkMySQLVarcharColumn(db, "shopee_products", "deleted_by_user_id", 191, true, "utf8mb4_bin", ""); err != nil { return err } for _, item := range []struct{ kind, name string }{{"index", "idx_shopee_products_deleted"}, {"constraint", "fk_shopee_products_deleted_by"}} { var exists bool var err error if item.kind == "index" { exists, err = mysqlIndexExists(db, "shopee_products", item.name) } else { exists, err = mysqlConstraintExists(db, "shopee_products", item.name) } if err != nil { return err } if !exists { return fmt.Errorf("蝦皮商品软删除%s %s 缺失", item.kind, item.name) } } return nil } func checkMySQLV18Shape(db *sql.DB) error { if err := checkMySQLSchema(db, []string{"shops", "shop_channel_aliases"}); err != nil { return err } for _, column := range []struct{ table, name string }{{"syb_orders", "shop_id"}, {"shopee_products", "shop_id"}} { if err := checkMySQLVarcharColumn(db, column.table, column.name, 191, true, "utf8mb4_bin", ""); err != nil { return err } } for _, item := range []struct{ table, name string }{ {"shops", "uq_shops_name"}, {"shops", "idx_shops_enabled"}, {"shop_channel_aliases", "uq_shop_channel_alias"}, {"shop_channel_aliases", "idx_shop_alias_shop"}, {"syb_orders", "idx_syb_orders_shop_id"}, {"shopee_products", "idx_shopee_products_shop_id"}, } { exists, err := mysqlIndexExists(db, item.table, item.name) if err != nil || !exists { return fmt.Errorf("店铺索引 %s.%s 缺失: %v", item.table, item.name, err) } } for _, item := range []struct{ table, name string }{ {"shops", "fk_shops_user"}, {"shops", "chk_shops_enabled"}, {"shop_channel_aliases", "fk_shop_alias_shop"}, {"shop_channel_aliases", "chk_shop_alias_channel"}, {"shop_channel_aliases", "chk_shop_alias_enabled"}, {"syb_orders", "fk_syb_orders_shop"}, {"shopee_products", "fk_shopee_products_shop"}, } { exists, err := mysqlConstraintExists(db, item.table, item.name) if err != nil || !exists { return fmt.Errorf("店铺约束 %s.%s 缺失: %v", item.table, item.name, err) } } return nil } func checkMySQLV19Shape(db *sql.DB) error { var invalid int err := db.QueryRow(`SELECT COUNT(*) FROM shops s WHERE (SELECT COUNT(*) FROM shop_channel_aliases a WHERE a.shop_id=s.shop_id AND a.channel='syb')<>1 OR (SELECT COUNT(*) FROM shop_channel_aliases a WHERE a.shop_id=s.shop_id AND a.channel='shopee')<>1 OR EXISTS (SELECT 1 FROM shop_channel_aliases a WHERE a.shop_id=s.shop_id AND (a.alias_name<>s.display_name OR a.normalized_alias<>s.normalized_name OR a.enabled<>s.enabled))`).Scan(&invalid) if err != nil { return fmt.Errorf("检查单一店铺名称兼容数据失败: %w", err) } if invalid != 0 { return fmt.Errorf("有 %d 个店铺的渠道兼容数据与唯一店铺名称不一致", invalid) } return nil } func checkMySQLV14Shape(db *sql.DB) error { if err := checkMySQLSchema(db, []string{"syb_allowed_shops"}); err != nil { return err } for _, column := range []struct { table, name string length int64 nullable bool collation, defVal string }{ {"syb_allowed_shops", "shop_id", 191, false, "utf8mb4_bin", ""}, {"syb_allowed_shops", "shop_name", 500, false, "utf8mb4_0900_ai_ci", ""}, {"syb_allowed_shops", "normalized_name", 500, false, "utf8mb4_bin", ""}, {"syb_allowed_shops", "created_by_user_id", 191, false, "utf8mb4_bin", ""}, {"syb_sync_runs", "shop_filter_hash", 64, true, "utf8mb4_bin", ""}, } { if err := checkMySQLVarcharColumn(db, column.table, column.name, column.length, column.nullable, column.collation, column.defVal); err != nil { return err } } for _, name := range []string{"accepted_stock_count", "shop_skipped_count"} { exists, err := mysqlColumnExists(db, "syb_sync_runs", name) if err != nil || !exists { return fmt.Errorf("顺运宝同步记录字段 %s 缺失", name) } } for _, name := range []string{"uq_syb_allowed_shops_name", "idx_syb_allowed_shops_enabled"} { exists, err := mysqlIndexExists(db, "syb_allowed_shops", name) if err != nil || !exists { return fmt.Errorf("顺运宝允许店铺索引 %s 缺失", name) } } for _, name := range []string{"fk_syb_allowed_shops_user", "chk_syb_allowed_shops_enabled"} { exists, err := mysqlConstraintExists(db, "syb_allowed_shops", name) if err != nil || !exists { return fmt.Errorf("顺运宝允许店铺约束 %s 缺失", name) } } return nil } func checkMySQLV9Shape(db *sql.DB) error { for _, column := range []struct { name string length int64 nullable bool collation, def string }{ {"image_url", 2048, true, "utf8mb4_0900_ai_ci", ""}, {"shopee_shop_name", 500, true, "utf8mb4_0900_ai_ci", ""}, {"image_source", 64, true, "utf8mb4_bin", ""}, {"image_observed_at", 35, true, "utf8mb4_0900_ai_ci", ""}, {"shop_name_source", 64, true, "utf8mb4_bin", ""}, {"shop_name_observed_at", 35, true, "utf8mb4_0900_ai_ci", ""}, } { if err := checkMySQLVarcharColumn(db, "shopee_products", column.name, column.length, column.nullable, column.collation, column.def); err != nil { return err } } for _, name := range []string{"image_is_manual", "shop_name_is_manual"} { var dataType, isNullable string var defaultValue sql.NullString err := db.QueryRow(`SELECT data_type,is_nullable,column_default FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='shopee_products' AND column_name=?`, name).Scan(&dataType, &isNullable, &defaultValue) if err != nil || dataType != "tinyint" || isNullable != "NO" || !defaultValue.Valid || defaultValue.String != "0" { return fmt.Errorf("蝦皮商品字段 %s 形状不正确", name) } } for _, name := range []string{"chk_shopee_products_image_manual", "chk_shopee_products_shop_manual"} { exists, err := mysqlConstraintExists(db, "shopee_products", name) if err != nil || !exists { return fmt.Errorf("蝦皮商品约束 %s 缺失", name) } } for _, name := range []string{"shopee_fields_filled", "shopee_fields_same_source_updated", "shopee_fields_manual_skipped", "shopee_fields_stale_skipped"} { exists, err := mysqlColumnExists(db, "catalog_import_runs", name) if err != nil || !exists { return fmt.Errorf("目录记录字段 %s 缺失", name) } } return nil } func checkMySQLV8Shape(db *sql.DB) error { for _, column := range []struct { name string length int64 nullable bool collation, def string }{ {"shopee_sku_id", 191, true, "utf8mb4_bin", ""}, {"spec_key", 191, true, "utf8mb4_bin", ""}, {"source", 64, true, "utf8mb4_bin", ""}, } { if err := checkMySQLVarcharColumn(db, "shopee_skus", column.name, column.length, column.nullable, column.collation, column.def); err != nil { return err } } for _, name := range []string{"field_sources", "field_observed_at"} { exists, err := mysqlColumnExists(db, "shopee_skus", name) if err != nil || !exists { return fmt.Errorf("蝦皮 SKU 字段 %s 缺失", name) } } if err := checkMySQLVarcharColumn(db, "catalog_import_runs", "update_policy", 32, false, "utf8mb4_bin", "fill_missing"); err != nil { return err } for _, name := range []string{"sku_filled", "sku_same_source_updated", "sku_skipped", "sku_manual_skipped", "sku_stale_skipped"} { exists, err := mysqlColumnExists(db, "catalog_import_runs", name) if err != nil || !exists { return fmt.Errorf("商品目录导入记录列 %s 缺失", name) } } for _, name := range []string{"uq_shopee_skus_external", "uq_shopee_skus_spec"} { exists, err := mysqlIndexExists(db, "shopee_skus", name) if err != nil || !exists { return fmt.Errorf("蝦皮 SKU 唯一索引 %s 缺失", name) } } return nil } func checkMySQLV7Shape(db *sql.DB) error { for _, column := range []struct{ table, name string }{ {"shopee_products", "source_observed_at"}, {"shopee_skus", "source_observed_at"}, {"pdd_products", "source_observed_at"}, } { if err := checkMySQLVarcharColumn(db, column.table, column.name, 35, true, "utf8mb4_0900_ai_ci", ""); err != nil { return err } if err := checkMySQLNullDefault(db, column.table, column.name); err != nil { return err } } if err := checkMySQLVarcharColumn(db, "pdd_products", "source", 16, true, "utf8mb4_bin", ""); err != nil { return err } if err := checkMySQLNullDefault(db, "pdd_products", "source"); err != nil { return err } if err := checkMySQLSchema(db, []string{"catalog_import_runs"}); err != nil { return err } for _, column := range []struct { name string length int64 nullable bool collation string def string }{ {"source", 64, false, "utf8mb4_bin", ""}, {"batch_id", 191, false, "utf8mb4_bin", ""}, {"request_hash", 64, false, "utf8mb4_bin", ""}, {"status", 16, false, "utf8mb4_bin", "processing"}, } { if err := checkMySQLVarcharColumn(db, "catalog_import_runs", column.name, column.length, column.nullable, column.collation, column.def); err != nil { return err } } var clause string if err := db.QueryRow(`SELECT cc.check_clause FROM information_schema.table_constraints tc JOIN information_schema.check_constraints cc ON cc.constraint_schema=tc.constraint_schema AND cc.constraint_name=tc.constraint_name WHERE tc.constraint_schema=DATABASE() AND tc.table_name='catalog_import_runs' AND tc.constraint_name='chk_catalog_runs_status' AND tc.constraint_type='CHECK'`).Scan(&clause); err != nil { return fmt.Errorf("商品目录批次状态约束缺失: %w", err) } normalized := strings.NewReplacer("`", "", " ", "", "(", "", ")", "", "_utf8mb4", "", `\`, "").Replace(strings.ToLower(clause)) for _, value := range []string{"'processing'", "'succeeded'", "'failed'"} { if !strings.Contains(normalized, value) { return fmt.Errorf("商品目录批次状态约束不正确") } } var sourceClause string if err := db.QueryRow(`SELECT cc.check_clause FROM information_schema.table_constraints tc JOIN information_schema.check_constraints cc ON cc.constraint_schema=tc.constraint_schema AND cc.constraint_name=tc.constraint_name WHERE tc.constraint_schema=DATABASE() AND tc.table_name='shopee_products' AND tc.constraint_name='chk_shopee_products_source' AND tc.constraint_type='CHECK'`).Scan(&sourceClause); err != nil || !strings.Contains(strings.ToLower(sourceClause), "api") { return fmt.Errorf("蝦皮商品来源约束未允许 api") } return nil } func checkMySQLV6Shape(db *sql.DB) error { if err := checkMySQLVarcharColumn(db, "tasks", "created_by_user_id", 191, true, "utf8mb4_bin", ""); err != nil { return err } if err := checkMySQLNullDefault(db, "tasks", "created_by_user_id"); err != nil { return err } var referencedTable, referencedColumn string if err := db.QueryRow(`SELECT referenced_table_name, referenced_column_name FROM information_schema.key_column_usage WHERE constraint_schema=DATABASE() AND table_name='tasks' AND constraint_name='fk_tasks_created_by' AND column_name='created_by_user_id'`).Scan(&referencedTable, &referencedColumn); err != nil || referencedTable != "users" || referencedColumn != "user_id" { return fmt.Errorf("任务创建人外键不正确") } var cols string if err := db.QueryRow(`SELECT GROUP_CONCAT(CONCAT(column_name,':',collation) ORDER BY seq_in_index) FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='tasks' AND index_name='idx_tasks_creator_list'`).Scan(&cols); err != nil || cols != "created_by_user_id:A,updated_at:D,task_id:D" { return fmt.Errorf("任务创建人列表索引不正确") } return nil } func checkMySQLV5Shape(db *sql.DB) error { if err := checkMySQLVarcharColumn(db, "tasks", "execution_mode", 16, false, "utf8mb4_bin", "dry_run"); err != nil { return err } if err := checkMySQLVarcharColumn(db, "tasks", "live_confirmed_by", 191, true, "utf8mb4_bin", ""); err != nil { return err } if err := checkMySQLNullDefault(db, "tasks", "live_confirmed_by"); err != nil { return err } if err := checkMySQLVarcharColumn(db, "tasks", "live_confirmed_at", 35, true, "utf8mb4_0900_ai_ci", ""); err != nil { return err } if err := checkMySQLNullDefault(db, "tasks", "live_confirmed_at"); err != nil { return err } for _, check := range []struct { name string required []string }{ {"chk_tasks_execution_mode", []string{"execution_modein'dry_run','live'"}}, {"chk_tasks_live_confirmation", []string{"execution_mode='dry_run'", "live_confirmed_byisnull", "live_confirmed_atisnull", "execution_mode='live'", "live_confirmed_byisnotnull", "live_confirmed_by<>''", "live_confirmed_atisnotnull", "live_confirmed_at<>''"}}, } { var enforced, clause string if err := db.QueryRow(`SELECT tc.enforced,cc.check_clause FROM information_schema.table_constraints tc JOIN information_schema.check_constraints cc ON cc.constraint_schema=tc.constraint_schema AND cc.constraint_name=tc.constraint_name WHERE tc.constraint_schema=DATABASE() AND tc.table_name='tasks' AND tc.constraint_name=? AND tc.constraint_type='CHECK'`, check.name).Scan(&enforced, &clause); err != nil { return fmt.Errorf("任务 CHECK %s 缺失或不可读: %w", check.name, err) } normalized := strings.NewReplacer("`", "", " ", "", "(", "", ")", "", "_utf8mb4", "", `\`, "").Replace(strings.ToLower(clause)) if enforced != "YES" { return fmt.Errorf("任务 CHECK %s 未启用", check.name) } for _, fragment := range check.required { if !strings.Contains(normalized, fragment) { return fmt.Errorf("任务 CHECK %s 表达式不正确", check.name) } } } var cols string if err := db.QueryRow(`SELECT GROUP_CONCAT(CONCAT(column_name,':',collation) ORDER BY seq_in_index) FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='tasks' AND index_name='idx_tasks_claim_mode'`).Scan(&cols); err != nil || cols != "assigned_client:A,status:A,execution_mode:A,priority:D,created_at:A" { return fmt.Errorf("任务模式领取索引不正确") } return nil } func checkMySQLNullDefault(db *sql.DB, table, column string) error { var defaultValue sql.NullString if err := db.QueryRow(`SELECT column_default FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name=? AND column_name=?`, table, column).Scan(&defaultValue); err != nil { return fmt.Errorf("检查 MySQL 列 %s.%s 默认值失败: %w", table, column, err) } if defaultValue.Valid { return fmt.Errorf("MySQL 列 %s.%s 默认值应为 NULL", table, column) } return nil } func checkMySQLV4Shape(db *sql.DB) error { for _, c := range []struct { name string length int64 nullable bool coll string }{ {"shopee_goods_id", 191, false, "utf8mb4_bin"}, {"spec_key", 191, false, "utf8mb4_bin"}, {"pdd_goods_id", 191, false, "utf8mb4_bin"}, {"rules_version", 32, false, "utf8mb4_0900_ai_ci"}, {"suggested_option_key", 191, true, "utf8mb4_bin"}, {"chosen_option_key", 191, false, "utf8mb4_bin"}, } { if err := checkMySQLVarcharColumn(db, "spec_mapping_decisions", c.name, c.length, c.nullable, c.coll, ""); err != nil { return err } } var enforced, clause string if err := db.QueryRow(`SELECT tc.enforced,cc.check_clause FROM information_schema.table_constraints tc JOIN information_schema.check_constraints cc ON cc.constraint_schema=tc.constraint_schema AND cc.constraint_name=tc.constraint_name WHERE tc.constraint_schema=DATABASE() AND tc.table_name='spec_mapping_decisions' AND tc.constraint_name='chk_spec_mapping_decisions_accepted'`).Scan(&enforced, &clause); err != nil { return fmt.Errorf("审计 accepted CHECK 缺失: %w", err) } n := strings.NewReplacer("`", "", " ", "", "(", "", ")", "", "_utf8mb4", "", `\`, "").Replace(strings.ToLower(clause)) if enforced != "YES" || n != "acceptedin0,1" { return fmt.Errorf("审计 accepted CHECK 不正确") } var cols 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='spec_mapping_decisions' AND index_name='idx_decisions_mapping'`).Scan(&cols); err != nil || cols != "shopee_goods_id,spec_key,pdd_goods_id" { return fmt.Errorf("审计映射索引不正确") } return nil } 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 { return fmt.Errorf("读取 MySQL 数据表清单失败: %w", err) } defer rows.Close() existing := map[string]bool{} for rows.Next() { var name string if err := rows.Scan(&name); err != nil { return fmt.Errorf("读取 MySQL 数据表清单失败: %w", err) } existing[name] = true } if err := rows.Err(); err != nil { return err } var missing []string for _, table := range tables { if !existing[table] { missing = append(missing, table) } } if len(missing) > 0 { return fmt.Errorf("MySQL schema 不完整，缺少表 %s", strings.Join(missing, "、")) } for table, columns := range requiredColumns { for _, column := range columns { var count int if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = ? AND column_name = ?`, table, column).Scan(&count); err != nil { return fmt.Errorf("检查 MySQL 数据表 %s 的列失败: %w", table, err) } if count != 1 { return fmt.Errorf("MySQL 数据表 %s 缺少列 %s", table, column) } } } return nil } func checkMySQLV3Shape(db *sql.DB) error { rows, err := db.Query(`SELECT column_name FROM information_schema.key_column_usage WHERE table_schema=DATABASE() AND table_name='spec_mappings' AND constraint_name='PRIMARY' ORDER BY ordinal_position`) if err != nil { return fmt.Errorf("检查 spec_mappings 主键失败: %w", err) } var primary []string for rows.Next() { var column string if err := rows.Scan(&column); err != nil { rows.Close() return fmt.Errorf("检查 spec_mappings 主键失败: %w", err) } primary = append(primary, column) } rows.Close() if strings.Join(primary, ",") != "shopee_goods_id,spec_key,pdd_goods_id" { return fmt.Errorf("spec_mappings 主键不正确，实际为 (%s)", strings.Join(primary, ",")) } for _, column := range []string{"shopee_goods_id", "spec_key", "pdd_goods_id"} { if err := checkMySQLVarcharColumn(db, "spec_mappings", column, 191, false, "utf8mb4_bin", ""); err != nil { return err } } if err := checkMySQLVarcharColumn(db, "syb_orders", "spec_key", 191, true, "utf8mb4_bin", ""); err != nil { return err } if err := checkMySQLVarcharColumn(db, "shopee_products", "source", 16, false, "utf8mb4_bin", "report"); err != nil { return err } var enforced, checkClause string if err := db.QueryRow(`SELECT tc.enforced, cc.check_clause FROM information_schema.table_constraints tc JOIN information_schema.check_constraints cc ON cc.constraint_schema=tc.constraint_schema AND cc.constraint_name=tc.constraint_name WHERE tc.constraint_schema=DATABASE() AND tc.table_name='shopee_products' AND tc.constraint_name='chk_shopee_products_source' AND tc.constraint_type='CHECK'`).Scan(&enforced, &checkClause); err != nil { return fmt.Errorf("shopee_products.source CHECK 缺失或不可读: %w", err) } if enforced != "YES" { return fmt.Errorf("shopee_products.source CHECK 未启用") } normalizedClause := strings.NewReplacer("`", "", " ", "", "(", "", ")", "", "_utf8mb4", "", `\`, "").Replace(strings.ToLower(checkClause)) if !strings.Contains(normalizedClause, "sourcein'report','syb'") { return fmt.Errorf("shopee_products.source CHECK 表达式不正确") } return nil } func checkMySQLVarcharColumn(db *sql.DB, table, column string, length int64, nullable bool, collation, defaultValue string) error { var dataType, isNullable string var actualLength sql.NullInt64 var actualCollation, actualDefault sql.NullString if err := db.QueryRow(`SELECT data_type, character_maximum_length, is_nullable, collation_name, column_default FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name=? AND column_name=?`, table, column).Scan(&dataType, &actualLength, &isNullable, &actualCollation, &actualDefault); err != nil { return fmt.Errorf("检查 MySQL 列 %s.%s 结构失败: %w", table, column, err) } wantNullable := "NO" if nullable { wantNullable = "YES" } if dataType != "varchar" || !actualLength.Valid || actualLength.Int64 != length || isNullable != wantNullable || actualCollation.String != collation { return fmt.Errorf("MySQL 列 %s.%s 结构不正确", table, column) } if defaultValue != "" && (!actualDefault.Valid || actualDefault.String != defaultValue) { return fmt.Errorf("MySQL 列 %s.%s 默认值不正确", table, column) } return nil }