feat: 建立商品目录接入基础 (#132)

admin/repository/catalog_import.go

@@ -0,0 +1,94 @@
+package repository
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/go-sql-driver/mysql"
+
+	"cmautobuy/admin/model"
+)
+
+var (
+	// ErrCatalogImportRunExists 表示同一来源的批次号已经登记，由 service 判断是否重放。
+	ErrCatalogImportRunExists   = errors.New("商品目录批次已经存在")
+	ErrCatalogImportRunNotFound = errors.New("商品目录批次不存在")
+)
+
+// InsertCatalogImportRun 先登记 processing 批次；表的复合主键是并发幂等的最终防线。
+func InsertCatalogImportRun(q Execer, run model.CatalogImportRun) error {
+	_, err := q.Exec(`INSERT INTO catalog_import_runs (
+		source,batch_id,request_hash,status,request_count,conflict_count,observed_at,
+		last_request_at,created_at
+	) VALUES (?,?,?,?,1,0,NULLIF(?,''),?,?)`, run.Source, run.BatchID, run.RequestHash,
+		run.Status, run.ObservedAt, run.LastRequestAt, run.CreatedAt)
+	if err == nil {
+		return nil
+	}
+	var mysqlErr *mysql.MySQLError
+	if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
+		return ErrCatalogImportRunExists
+	}
+	if strings.Contains(strings.ToLower(err.Error()), "unique constraint failed") {
+		return ErrCatalogImportRunExists
+	}
+	return fmt.Errorf("登记商品目录批次失败: %w", err)
+}
+
+// GetCatalogImportRun 查询一个批次的摘要，不读取原始请求体。
+func GetCatalogImportRun(q Execer, source, batchID string) (model.CatalogImportRun, error) {
+	var run model.CatalogImportRun
+	var observedAt, lastConflictAt, errorSummary, responseBody, finishedAt sql.NullString
+	err := q.QueryRow(`SELECT source,batch_id,request_hash,status,request_count,conflict_count,
+		observed_at,last_request_at,last_conflict_at,shopee_created,shopee_updated,sku_created,
+		sku_updated,pdd_created,pdd_updated,association_created,association_unchanged,failure_count,
+		error_summary,response_body,created_at,finished_at
+		FROM catalog_import_runs WHERE source=? AND batch_id=?`, source, batchID).Scan(
+		&run.Source, &run.BatchID, &run.RequestHash, &run.Status, &run.RequestCount, &run.ConflictCount,
+		&observedAt, &run.LastRequestAt, &lastConflictAt, &run.ShopeeCreated, &run.ShopeeUpdated,
+		&run.SKUCreated, &run.SKUUpdated, &run.PddCreated, &run.PddUpdated, &run.AssociationCreated,
+		&run.AssociationUnchanged, &run.FailureCount, &errorSummary, &responseBody, &run.CreatedAt, &finishedAt)
+	if errors.Is(err, sql.ErrNoRows) {
+		return model.CatalogImportRun{}, ErrCatalogImportRunNotFound
+	}
+	if err != nil {
+		return model.CatalogImportRun{}, fmt.Errorf("查询商品目录批次失败: %w", err)
+	}
+	run.ObservedAt = observedAt.String
+	run.LastConflictAt = lastConflictAt.String
+	run.ErrorSummary = errorSummary.String
+	run.ResponseBody = responseBody.String
+	run.FinishedAt = finishedAt.String
+	return run, nil
+}
+
+// RecordCatalogImportReplay 记录相同请求的重复提交，业务数据不会再次写入。
+func RecordCatalogImportReplay(q Execer, source, batchID, requestedAt string) error {
+	result, err := q.Exec(`UPDATE catalog_import_runs SET request_count=request_count+1,last_request_at=?
+		WHERE source=? AND batch_id=?`, requestedAt, source, batchID)
+	return catalogRunUpdateResult(result, err, "记录商品目录批次重放")
+}
+
+// RecordCatalogImportConflict 记录同一批次号携带不同内容的冲突。
+func RecordCatalogImportConflict(q Execer, source, batchID, requestedAt string) error {
+	result, err := q.Exec(`UPDATE catalog_import_runs SET request_count=request_count+1,
+		conflict_count=conflict_count+1,last_request_at=?,last_conflict_at=?
+		WHERE source=? AND batch_id=?`, requestedAt, requestedAt, source, batchID)
+	return catalogRunUpdateResult(result, err, "记录商品目录批次冲突")
+}
+
+func catalogRunUpdateResult(result sql.Result, err error, action string) error {
+	if err != nil {
+		return fmt.Errorf("%s失败: %w", action, err)
+	}
+	affected, err := result.RowsAffected()
+	if err != nil {
+		return fmt.Errorf("读取%s结果失败: %w", action, err)
+	}
+	if affected == 0 {
+		return ErrCatalogImportRunNotFound
+	}
+	return nil
+}

admin/repository/catalog_import_test.go

@@ -0,0 +1,52 @@
+package repository
+
+import (
+	"database/sql"
+	"errors"
+	"testing"
+
+	_ "modernc.org/sqlite"
+
+	"cmautobuy/admin/model"
+)
+
+func TestCatalogImportRun_登记查询重放和冲突(t *testing.T) {
+	db, err := sql.Open("sqlite", ":memory:")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.Exec(`CREATE TABLE catalog_import_runs (
+		source TEXT NOT NULL,batch_id TEXT NOT NULL,request_hash TEXT NOT NULL,status TEXT NOT NULL,
+		request_count INTEGER NOT NULL,conflict_count INTEGER NOT NULL,observed_at TEXT,last_request_at TEXT NOT NULL,
+		last_conflict_at TEXT,shopee_created INTEGER NOT NULL DEFAULT 0,shopee_updated INTEGER NOT NULL DEFAULT 0,
+		sku_created INTEGER NOT NULL DEFAULT 0,sku_updated INTEGER NOT NULL DEFAULT 0,pdd_created INTEGER NOT NULL DEFAULT 0,
+		pdd_updated INTEGER NOT NULL DEFAULT 0,association_created INTEGER NOT NULL DEFAULT 0,
+		association_unchanged INTEGER NOT NULL DEFAULT 0,failure_count INTEGER NOT NULL DEFAULT 0,
+		error_summary TEXT,response_body TEXT,created_at TEXT NOT NULL,finished_at TEXT,
+		PRIMARY KEY(source,batch_id))`); err != nil {
+		t.Fatal(err)
+	}
+	run := model.CatalogImportRun{Source: "script-a", BatchID: "batch-1", RequestHash: "abc",
+		Status: model.CatalogImportProcessing, ObservedAt: "2026-08-11T00:00:00Z",
+		LastRequestAt: "2026-08-11T00:01:00Z", CreatedAt: "2026-08-11T00:01:00Z"}
+	if err := InsertCatalogImportRun(db, run); err != nil {
+		t.Fatal(err)
+	}
+	if err := InsertCatalogImportRun(db, run); !errors.Is(err, ErrCatalogImportRunExists) {
+		t.Fatalf("重复批次应由唯一键拦截：%v", err)
+	}
+	if err := RecordCatalogImportReplay(db, run.Source, run.BatchID, "2026-08-11T00:02:00Z"); err != nil {
+		t.Fatal(err)
+	}
+	if err := RecordCatalogImportConflict(db, run.Source, run.BatchID, "2026-08-11T00:03:00Z"); err != nil {
+		t.Fatal(err)
+	}
+	got, err := GetCatalogImportRun(db, run.Source, run.BatchID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.RequestCount != 3 || got.ConflictCount != 1 || got.LastConflictAt == "" {
+		t.Fatalf("批次计数不正确：%+v", got)
+	}
+}

admin/repository/mysql_db.go

@@ -19,7 +19,7 @@ import (
 	"cmautobuy/admin/spec"
 )
 
-const mysqlSchemaVersion = 6
+const mysqlSchemaVersion = 7
 
 // OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。
 func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) {
@@ -495,10 +495,98 @@ func MigrateMySQL(db *sql.DB) error {
 		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)
+		}
 	}
 	return CheckMySQLSchema(db)
 }
 
+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 {
@@ -843,6 +931,7 @@ func CheckMySQLSchema(db *sql.DB) error {
 		"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",
 	}
 	if err := checkMySQLSchema(db, mysqlRequiredTables); err != nil {
 		return err
@@ -856,7 +945,71 @@ func CheckMySQLSchema(db *sql.DB) error {
 	if err := checkMySQLV5Shape(db); err != nil {
 		return err
 	}
-	return checkMySQLV6Shape(db)
+	if err := checkMySQLV6Shape(db); err != nil {
+		return err
+	}
+	return checkMySQLV7Shape(db)
+}
+
+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 {

admin/repository/mysql_db_integration_test.go

@@ -340,6 +340,48 @@ func TestMySQLMigrate_V6形状错误不记版本(t *testing.T) {
 	}
 }
 
+func TestMySQLMigrate_V6升级V7且断点重跑(t *testing.T) {
+	db := openMySQLMigrationTestDB(t)
+	defer db.Close()
+	cleanMySQLTestSchema(t, db)
+	defer cleanMySQLTestSchema(t, db)
+	prepareMySQLV6(t, db)
+
+	// 模拟 MySQL DDL 已经提交、版本号尚未记录的中断状态。
+	if err := migrateMySQLV7(db); err != nil {
+		t.Fatal(err)
+	}
+	if err := MigrateMySQL(db); err != nil {
+		t.Fatal(err)
+	}
+	if err := MigrateMySQL(db); err != nil {
+		t.Fatalf("v7 重跑失败: %v", err)
+	}
+	var versions int
+	if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=7`).Scan(&versions); err != nil || versions != 1 {
+		t.Fatalf("v7=%d err=%v", versions, err)
+	}
+	mustExec(t, db, `INSERT INTO shopee_products(goods_id,title,source,created_at,updated_at)
+		VALUES('API-1','接口商品','api','2026-08-11T00:00:00Z','2026-08-11T00:00:00Z')`)
+}
+
+func TestMySQLMigrate_V7形状错误不记版本(t *testing.T) {
+	db := openMySQLMigrationTestDB(t)
+	defer db.Close()
+	cleanMySQLTestSchema(t, db)
+	defer cleanMySQLTestSchema(t, db)
+	prepareMySQLV6(t, db)
+	mustExec(t, db, `ALTER TABLE shopee_products ADD COLUMN source_observed_at VARCHAR(10) NULL`)
+	if err := MigrateMySQL(db); err == nil {
+		t.Fatal("错误 source_observed_at 形状必须阻止 v7")
+	}
+	var count int
+	db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=7`).Scan(&count)
+	if count != 0 {
+		t.Fatal("v7 自检失败不得记录版本")
+	}
+}
+
 func openMySQLMigrationTestDB(t *testing.T) *sql.DB {
 	t.Helper()
 	if os.Getenv("CMAUTOBUY_MYSQL_TEST") != "1" {
@@ -381,6 +423,19 @@ func prepareMySQLV4(t *testing.T, db *sql.DB) {
 	mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (3,'2026-08-10T00:00:00Z'),(4,'2026-08-10T00:00:00Z')`)
 }
 
+func prepareMySQLV6(t *testing.T, db *sql.DB) {
+	t.Helper()
+	prepareMySQLV4(t, db)
+	if err := migrateMySQLV5(db); err != nil {
+		t.Fatal(err)
+	}
+	if err := migrateMySQLV6(db); err != nil {
+		t.Fatal(err)
+	}
+	mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at)
+		VALUES (5,'2026-08-10T00:00:00Z'),(6,'2026-08-10T00:00:00Z')`)
+}
+
 func mustExec(t *testing.T, db *sql.DB, query string, args ...any) {
 	t.Helper()
 	if _, err := db.Exec(query, args...); err != nil {