feat(distribution): 优化一键同步主站商品功能(并发防重、分批入库、失效商品联动下架与初始加价)
- 后端优化: - 引入 Redis + 内存 sync.Map 双重互斥锁,避免并发/连击导致唯一索引冲突 (ErrSyncInProgress) - 使用 CreateInBatches(500) 进行大数据量批量插入,杜绝超长 SQL 与锁表超时 - 增加 CleanStaleListings 支持,自动排查并联动下架主站非在售的货架商品,清理历史幽灵数据 - 增加 DefaultShelfRatioDelta 支持,新同步商品直接继承初始加价比例,免除二次手动改价 - 前端优化: - distribution-admin 货架管理新增一键同步配置弹窗(可勾选自动下架失效商品、配置初始加价) - 同步完成后展示结构化统计弹窗(新增、保持、联动下架、主站在售总量) - 前后端类型定义与 API 客户端对齐升级 - 测试覆盖: - 补充 TestSyncSiteAccountsOptimizations 单元测试与 handler 集成测试 - 全工程测试与门禁 check.sh 100% 通过
This commit is contained in:
@@ -9,6 +9,7 @@ import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"hfb_sys/backend/internal/auditlog"
|
||||
@@ -64,8 +65,9 @@ func normalizeSettlementHoldDays(days int) (int, error) {
|
||||
}
|
||||
|
||||
type Repository struct {
|
||||
db *gorm.DB
|
||||
redis *redis.Client
|
||||
db *gorm.DB
|
||||
redis *redis.Client
|
||||
siteSyncLocks sync.Map
|
||||
}
|
||||
|
||||
func NewRepository(db *gorm.DB, redis ...*redis.Client) *Repository {
|
||||
@@ -914,7 +916,7 @@ func uniquePositiveIDs(ids []uint64) []uint64 {
|
||||
return out
|
||||
}
|
||||
|
||||
func (r *Repository) SyncSiteAccounts(ctx context.Context, adminID, siteID uint64, listingIDs []uint64) (*SyncSiteAccountsResult, error) {
|
||||
func (r *Repository) SyncSiteAccounts(ctx context.Context, adminID, siteID uint64, req SyncSiteAccountsRequest) (*SyncSiteAccountsResult, error) {
|
||||
if r.db == nil {
|
||||
return nil, ErrDependencyUnavailable
|
||||
}
|
||||
@@ -929,7 +931,30 @@ func (r *Repository) SyncSiteAccounts(ctx context.Context, adminID, siteID uint6
|
||||
return nil, err
|
||||
}
|
||||
|
||||
selectedIDs := uniquePositiveIDs(listingIDs)
|
||||
if req.DefaultShelfRatioDelta < 0 {
|
||||
return nil, ErrInvalidAmount
|
||||
}
|
||||
|
||||
// 1. 防重复并发锁(Redis 分布式锁 + 内存 sync.Map 双重互斥保障)
|
||||
lockKey := fmt.Sprintf("dist:sync_accounts:%d", siteID)
|
||||
if r.redis != nil {
|
||||
ok, err := r.redis.SetNX(ctx, lockKey, 1, 30*time.Second).Result()
|
||||
if err == nil {
|
||||
if !ok {
|
||||
return nil, ErrSyncInProgress
|
||||
}
|
||||
defer func() {
|
||||
_ = r.redis.Del(context.Background(), lockKey).Err()
|
||||
}()
|
||||
}
|
||||
}
|
||||
_, loaded := r.siteSyncLocks.LoadOrStore(siteID, struct{}{})
|
||||
if loaded {
|
||||
return nil, ErrSyncInProgress
|
||||
}
|
||||
defer r.siteSyncLocks.Delete(siteID)
|
||||
|
||||
selectedIDs := uniquePositiveIDs(req.ListingIDs)
|
||||
if len(selectedIDs) > maxSyncListingIDs {
|
||||
return nil, ErrTooManySyncListings
|
||||
}
|
||||
@@ -969,6 +994,7 @@ func (r *Repository) SyncSiteAccounts(ctx context.Context, adminID, siteID uint6
|
||||
var toInsert []model.DistributionSiteAccount
|
||||
seenAccount := make(map[uint64]bool)
|
||||
var skippedCount int64
|
||||
var updatedCount int64
|
||||
|
||||
for _, l := range candidates {
|
||||
if seenAccount[l.AccountID] {
|
||||
@@ -980,28 +1006,55 @@ func (r *Repository) SyncSiteAccounts(ctx context.Context, adminID, siteID uint6
|
||||
if ea.ListingID != l.ID {
|
||||
_ = r.db.WithContext(ctx).Model(&model.DistributionSiteAccount{}).
|
||||
Where("id = ?", ea.ID).Update("listing_id", l.ID).Error
|
||||
updatedCount++
|
||||
}
|
||||
skippedCount++
|
||||
} else {
|
||||
toInsert = append(toInsert, model.DistributionSiteAccount{
|
||||
SiteID: siteID,
|
||||
AccountID: l.AccountID,
|
||||
ListingID: l.ID,
|
||||
DisplayStatus: "enabled",
|
||||
SiteID: siteID,
|
||||
AccountID: l.AccountID,
|
||||
ListingID: l.ID,
|
||||
DisplayStatus: "enabled",
|
||||
ShelfRatioDelta: req.DefaultShelfRatioDelta,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 2. 分批高效入库,防止大数据量超长 SQL 或锁表超时
|
||||
if len(toInsert) > 0 {
|
||||
if err := r.db.WithContext(ctx).Create(&toInsert).Error; err != nil {
|
||||
if err := r.db.WithContext(ctx).CreateInBatches(&toInsert, 500).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// 3. 联动下架主站非在售/已失效的货架商品(若开启 CleanStaleListings)
|
||||
var disabledStaleCount int64
|
||||
if req.CleanStaleListings {
|
||||
var staleIDs []uint64
|
||||
err := r.db.WithContext(ctx).Table("distribution_site_accounts AS dsa").
|
||||
Joins("LEFT JOIN rental_listings AS l ON l.id = dsa.listing_id AND l.status = 'published' AND l.review_status = 'approved'").
|
||||
Where("dsa.site_id = ? AND dsa.display_status = 'enabled' AND l.id IS NULL", siteID).
|
||||
Pluck("dsa.id", &staleIDs).Error
|
||||
if err == nil && len(staleIDs) > 0 {
|
||||
for i := 0; i < len(staleIDs); i += 500 {
|
||||
end := i + 500
|
||||
if end > len(staleIDs) {
|
||||
end = len(staleIDs)
|
||||
}
|
||||
_ = r.db.WithContext(ctx).Model(&model.DistributionSiteAccount{}).
|
||||
Where("id IN ?", staleIDs[i:end]).
|
||||
Update("display_status", "disabled").Error
|
||||
}
|
||||
disabledStaleCount = int64(len(staleIDs))
|
||||
}
|
||||
}
|
||||
|
||||
return &SyncSiteAccountsResult{
|
||||
SyncedCount: int64(len(toInsert)),
|
||||
SkippedCount: skippedCount,
|
||||
TotalActive: int64(len(candidates)),
|
||||
SyncedCount: int64(len(toInsert)),
|
||||
SkippedCount: skippedCount,
|
||||
UpdatedCount: updatedCount,
|
||||
DisabledStaleCount: disabledStaleCount,
|
||||
TotalActive: int64(len(candidates)),
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user