cmautobuy/admin/repository/catalog_import.go

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
}

// ListCatalogImportRuns 在数据库中按来源、状态筛选并倒序分页。
func ListCatalogImportRuns(q Execer, source string, status model.CatalogImportStatus, limit, offset int) ([]model.CatalogImportRun, int, error) {
	where := " WHERE 1=1"
	args := make([]any, 0, 4)
	if source != "" {
		where += " AND source=?"
		args = append(args, source)
	}
	if status != "" {
		where += " AND status=?"
		args = append(args, status)
	}
	var total int
	if err := q.QueryRow(`SELECT COUNT(*) FROM catalog_import_runs`+where, args...).Scan(&total); err != nil {
		return nil, 0, fmt.Errorf("统计商品目录批次失败: %w", err)
	}
	listArgs := append(append([]any{}, args...), limit, offset)
	rows, err := q.Query(`SELECT source,batch_id FROM catalog_import_runs`+where+` ORDER BY created_at DESC,source,batch_id LIMIT ? OFFSET ?`, listArgs...)
	if err != nil {
		return nil, 0, fmt.Errorf("查询商品目录批次列表失败: %w", err)
	}
	defer rows.Close()
	keys := make([][2]string, 0, limit)
	for rows.Next() {
		var key [2]string
		if err := rows.Scan(&key[0], &key[1]); err != nil {
			return nil, 0, err
		}
		keys = append(keys, key)
	}
	if err := rows.Err(); err != nil {
		return nil, 0, err
	}
	if err := rows.Close(); err != nil {
		return nil, 0, err
	}
	list := make([]model.CatalogImportRun, 0, len(keys))
	for _, key := range keys {
		run, err := GetCatalogImportRun(q, key[0], key[1])
		if err != nil {
			return nil, 0, err
		}
		list = append(list, run)
	}
	return list, total, nil
}

// ListCatalogImportSources 返回筛选下拉框所需的稳定来源名。
func ListCatalogImportSources(q Execer) ([]string, error) {
	rows, err := q.Query(`SELECT DISTINCT source FROM catalog_import_runs ORDER BY source`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var list []string
	for rows.Next() {
		var source string
		if err := rows.Scan(&source); err != nil {
			return nil, err
		}
		list = append(list, source)
	}
	return list, rows.Err()
}

// 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, "记录商品目录批次冲突")
}

// CompleteCatalogImportRun 把处理结果摘要和安全的响应 JSON 固化，供幂等重放。
func CompleteCatalogImportRun(q Execer, run model.CatalogImportRun) error {
	result, err := q.Exec(`UPDATE catalog_import_runs SET status=?,shopee_created=?,shopee_updated=?,
		sku_created=?,sku_updated=?,pdd_created=?,pdd_updated=?,association_created=?,
		association_unchanged=?,failure_count=?,error_summary=NULLIF(?,''),response_body=NULLIF(?,''),
		finished_at=? WHERE source=? AND batch_id=?`, run.Status, run.ShopeeCreated, run.ShopeeUpdated,
		run.SKUCreated, run.SKUUpdated, run.PddCreated, run.PddUpdated, run.AssociationCreated,
		run.AssociationUnchanged, run.FailureCount, run.ErrorSummary, run.ResponseBody,
		run.FinishedAt, run.Source, run.BatchID)
	return catalogRunUpdateResult(result, err, "完成商品目录批次")
}

// UpsertCatalogShopeeProduct 按上游观测时间更新来源字段，绝不覆盖人工 PDD 关联。
func UpsertCatalogShopeeProduct(q Execer, goodsID, title, status, mainSKU, observedAt, now string) (created, updated bool, err error) {
	var oldObserved sql.NullString
	err = q.QueryRow(`SELECT source_observed_at FROM shopee_products WHERE goods_id=?`, goodsID).Scan(&oldObserved)
	if errors.Is(err, sql.ErrNoRows) {
		_, err = q.Exec(`INSERT INTO shopee_products(goods_id,title,shopee_status,main_sku_code,source,
			source_observed_at,created_at,updated_at) VALUES(?,?,NULLIF(?,''),NULLIF(?,''),'api',?,?,?)`,
			goodsID, title, status, mainSKU, observedAt, now, now)
		return err == nil, false, err
	}
	if err != nil {
		return false, false, err
	}
	if oldObserved.Valid && oldObserved.String > observedAt {
		return false, false, nil
	}
	_, err = q.Exec(`UPDATE shopee_products SET title=?,shopee_status=NULLIF(?,''),main_sku_code=NULLIF(?,''),
		source='api',source_observed_at=?,updated_at=? WHERE goods_id=?`, title, status, mainSKU, observedAt, now, goodsID)
	return false, err == nil, err
}

// UpsertCatalogShopeeSKU 拒绝把同一 SKU 静默挪到另一商品，并保留 is_manual。
func UpsertCatalogShopeeSKU(q Execer, skuID, goodsID, specRaw, color, size, advice string, parseOK bool, skuCode, observedAt, now string) (created, updated bool, err error) {
	var oldGoods string
	var oldObserved sql.NullString
	err = q.QueryRow(`SELECT goods_id,source_observed_at FROM shopee_skus WHERE sku_id=?`, skuID).Scan(&oldGoods, &oldObserved)
	parse := 0
	if parseOK {
		parse = 1
	}
	if errors.Is(err, sql.ErrNoRows) {
		_, err = q.Exec(`INSERT INTO shopee_skus(sku_id,goods_id,spec_raw,color,size,advice,parse_ok,sku_code,
			is_manual,source_observed_at,created_at,updated_at) VALUES(?,?,?,NULLIF(?,''),NULLIF(?,''),NULLIF(?,''),?,NULLIF(?,''),0,?,?,?)`,
			skuID, goodsID, specRaw, color, size, advice, parse, skuCode, observedAt, now, now)
		return err == nil, false, err
	}
	if err != nil {
		return false, false, err
	}
	if oldGoods != goodsID {
		return false, false, fmt.Errorf("SKU %s 已属于蝦皮商品 %s", skuID, oldGoods)
	}
	if oldObserved.Valid && oldObserved.String > observedAt {
		return false, false, nil
	}
	_, err = q.Exec(`UPDATE shopee_skus SET spec_raw=?,color=NULLIF(?,''),size=NULLIF(?,''),advice=NULLIF(?,''),
		parse_ok=?,sku_code=NULLIF(?,''),source_observed_at=?,updated_at=? WHERE sku_id=?`,
		specRaw, color, size, advice, parse, skuCode, observedAt, now, skuID)
	return false, err == nil, err
}

// UpsertCatalogPddProduct 写入结构化 PDD 数据转换后的规范 JSON；空规格不清除既有采集结果。
func UpsertCatalogPddProduct(q Execer, goodsID, url, title, shopName, skusJSON, observedAt, now string) (created, updated bool, err error) {
	var oldObserved, deletedAt sql.NullString
	err = q.QueryRow(`SELECT source_observed_at,deleted_at FROM pdd_products WHERE goods_id=?`, goodsID).Scan(&oldObserved, &deletedAt)
	status := "pending"
	if skusJSON != "" {
		status = "collected"
	}
	if errors.Is(err, sql.ErrNoRows) {
		_, err = q.Exec(`INSERT INTO pdd_products(goods_id,url,title,shop_name,skus_json,collect_status,collected_at,
			source,source_observed_at,created_at,updated_at) VALUES(?,?,NULLIF(?,''),NULLIF(?,''),NULLIF(?,''),?,
			CASE WHEN ?='collected' THEN ? ELSE NULL END,'api',?,?,?)`, goodsID, url, title, shopName, skusJSON, status, status, now, observedAt, now, now)
		return err == nil, false, err
	}
	if err != nil {
		return false, false, err
	}
	if oldObserved.Valid && oldObserved.String > observedAt {
		return false, false, nil
	}
	if skusJSON == "" {
		_, err = q.Exec(`UPDATE pdd_products SET url=?,title=CASE WHEN TRIM(?)='' THEN title ELSE ? END,
			shop_name=CASE WHEN TRIM(?)='' THEN shop_name ELSE ? END,deleted_at=NULL,source='api',source_observed_at=?,updated_at=? WHERE goods_id=?`,
			url, title, title, shopName, shopName, observedAt, now, goodsID)
	} else {
		_, err = q.Exec(`UPDATE pdd_products SET url=?,title=NULLIF(?,''),shop_name=NULLIF(?,''),skus_json=?,
			collect_status='collected',collect_msg=NULL,artifact_ref=NULL,collected_at=?,deleted_at=NULL,
			source='api',source_observed_at=?,updated_at=? WHERE goods_id=?`, url, title, shopName, skusJSON, now, observedAt, now, goodsID)
	}
	return false, err == nil, err
}

// ApplyCatalogAssociation 只允许空关联建立或相同关联重放。
func ApplyCatalogAssociation(q Execer, shopeeGoodsID, pddGoodsID, now string) (created, unchanged bool, err error) {
	var current sql.NullString
	if err = q.QueryRow(`SELECT pdd_goods_id FROM shopee_products WHERE goods_id=?`, shopeeGoodsID).Scan(&current); err != nil {
		return
	}
	var url string
	if err = q.QueryRow(`SELECT url FROM pdd_products WHERE goods_id=? AND deleted_at IS NULL`, pddGoodsID).Scan(&url); err != nil {
		return
	}
	if current.Valid && current.String != "" {
		if current.String == pddGoodsID {
			return false, true, nil
		}
		return false, false, fmt.Errorf("蝦皮商品 %s 已关联 PDD 商品 %s", shopeeGoodsID, current.String)
	}
	_, err = q.Exec(`UPDATE shopee_products SET pdd_goods_id=?,pdd_goods_url=?,updated_at=? WHERE goods_id=?`, pddGoodsID, url, now, shopeeGoodsID)
	return err == nil, false, 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
}