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 }