From 8f738ffbe85abfc8cc3f046b719b6dfe4f4bdff5 Mon Sep 17 00:00:00 2001 From: break Date: Thu, 2 Jul 2026 13:04:36 +0900 Subject: [PATCH] =?UTF-8?q?=E5=AF=BC=E5=85=A5=E7=9A=84=E9=97=AE=E9=A2=98?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../postgres/device_import_task_store.go | 10 ++ .../postgres/iot_card_import_task_store.go | 10 ++ internal/task/device_import.go | 2 + internal/task/iot_card_import.go | 132 +++++++++++------- pkg/queue/handler.go | 1 - 5 files changed, 101 insertions(+), 54 deletions(-) diff --git a/internal/store/postgres/device_import_task_store.go b/internal/store/postgres/device_import_task_store.go index 3de86e5..274bf78 100644 --- a/internal/store/postgres/device_import_task_store.go +++ b/internal/store/postgres/device_import_task_store.go @@ -67,6 +67,16 @@ func (s *DeviceImportTaskStore) UpdateStatus(ctx context.Context, id uint, statu return s.db.WithContext(ctx).Model(&model.DeviceImportTask{}).Where("id = ?", id).Updates(updates).Error } +// UpdateProgress 实时更新导入进度计数（不更新详情列表，用于批量处理中间状态） +func (s *DeviceImportTaskStore) UpdateProgress(ctx context.Context, id uint, successCount, skipCount, failCount int) error { + return s.db.WithContext(ctx).Model(&model.DeviceImportTask{}).Where("id = ?", id).Updates(map[string]any{ + "success_count": successCount, + "skip_count": skipCount, + "fail_count": failCount, + "updated_at": time.Now(), + }).Error +} + func (s *DeviceImportTaskStore) UpdateResult(ctx context.Context, id uint, totalCount, successCount, skipCount, failCount, warningCount int, skippedItems, failedItems, warningItems model.ImportResultItems) error { updates := map[string]any{ "total_count": totalCount, diff --git a/internal/store/postgres/iot_card_import_task_store.go b/internal/store/postgres/iot_card_import_task_store.go index 3226c3a..28f2642 100644 --- a/internal/store/postgres/iot_card_import_task_store.go +++ b/internal/store/postgres/iot_card_import_task_store.go @@ -72,6 +72,16 @@ func (s *IotCardImportTaskStore) UpdateStatus(ctx context.Context, id uint, stat return s.db.WithContext(ctx).Model(&model.IotCardImportTask{}).Where("id = ?", id).Updates(updates).Error } +// UpdateProgress 实时更新导入进度计数（不更新详情列表，用于批量处理中间状态） +func (s *IotCardImportTaskStore) UpdateProgress(ctx context.Context, id uint, successCount, skipCount, failCount int) error { + return s.db.WithContext(ctx).Model(&model.IotCardImportTask{}).Where("id = ?", id).Updates(map[string]interface{}{ + "success_count": successCount, + "skip_count": skipCount, + "fail_count": failCount, + "updated_at": time.Now(), + }).Error +} + func (s *IotCardImportTaskStore) UpdateResult(ctx context.Context, id uint, successCount, skipCount, failCount int, skippedItems, failedItems model.ImportResultItems) error { updates := map[string]interface{}{ "success_count": successCount, diff --git a/internal/task/device_import.go b/internal/task/device_import.go index efc332d..0ac6591 100644 --- a/internal/task/device_import.go +++ b/internal/task/device_import.go @@ -185,6 +185,8 @@ func (h *DeviceImportHandler) processImport(ctx context.Context, task *model.Dev end := min(i+deviceBatchSize, len(rows)) batch := rows[i:end] h.processBatch(ctx, task, batch, result) + // 每批完成后实时更新进度计数，让前端可以看到处理进度 + h.importTaskStore.UpdateProgress(ctx, task.ID, result.successCount, result.skipCount, result.failCount) } return result diff --git a/internal/task/iot_card_import.go b/internal/task/iot_card_import.go index 4297489..84fd9f1 100644 --- a/internal/task/iot_card_import.go +++ b/internal/task/iot_card_import.go @@ -41,15 +41,14 @@ type PollingCallback interface { } type IotCardImportHandler struct { - db *gorm.DB - redis *redis.Client - importTaskStore *postgres.IotCardImportTaskStore - iotCardStore *postgres.IotCardStore - assetWalletStore *postgres.AssetWalletStore - assetIdentifierStore *postgres.AssetIdentifierStore - storageService *storage.Service - pollingCallback PollingCallback - logger *zap.Logger + db *gorm.DB + redis *redis.Client + importTaskStore *postgres.IotCardImportTaskStore + iotCardStore *postgres.IotCardStore + assetWalletStore *postgres.AssetWalletStore + storageService *storage.Service + pollingCallback PollingCallback + logger *zap.Logger } func NewIotCardImportHandler( @@ -58,21 +57,19 @@ func NewIotCardImportHandler( importTaskStore *postgres.IotCardImportTaskStore, iotCardStore *postgres.IotCardStore, assetWalletStore *postgres.AssetWalletStore, - assetIdentifierStore *postgres.AssetIdentifierStore, storageSvc *storage.Service, pollingCallback PollingCallback, logger *zap.Logger, ) *IotCardImportHandler { return &IotCardImportHandler{ - db: db, - redis: redis, - importTaskStore: importTaskStore, - iotCardStore: iotCardStore, - assetWalletStore: assetWalletStore, - assetIdentifierStore: assetIdentifierStore, - storageService: storageSvc, - pollingCallback: pollingCallback, - logger: logger, + db: db, + redis: redis, + importTaskStore: importTaskStore, + iotCardStore: iotCardStore, + assetWalletStore: assetWalletStore, + storageService: storageSvc, + pollingCallback: pollingCallback, + logger: logger, } } @@ -218,6 +215,8 @@ func (h *IotCardImportHandler) processImport(ctx context.Context, task *model.Io end := min(i+batchSize, len(cards)) batch := cards[i:end] h.processBatch(ctx, task, batch, i+1, result) + // 每批完成后实时更新进度计数，让前端可以看到处理进度 + h.importTaskStore.UpdateProgress(ctx, task.ID, result.successCount, result.skipCount, result.failCount) } return result @@ -341,7 +340,14 @@ func (h *IotCardImportHandler) processBatch(ctx context.Context, task *model.Iot } now := time.Now() - createdCards := make([]*model.IotCard, 0, len(finalCards)) + + // 构建待批量写入的卡列表，同时记录原始信息用于失败回溯 + type pendingEntry struct { + item model.CardItem + meta cardMeta + } + pending := make([]pendingEntry, 0, len(finalCards)) + iotCards := make([]*model.IotCard, 0, len(finalCards)) for _, card := range finalCards { meta := cardMetaMap[card.ICCID] @@ -386,43 +392,63 @@ func (h *IotCardImportHandler) processBatch(ctx context.Context, task *model.Iot iotCard.CreatedAt = now iotCard.UpdatedAt = now - txErr := h.db.Transaction(func(tx *gorm.DB) error { - if err := tx.Create(iotCard).Error; err != nil { - return err - } - txIdentifierStore := postgres.NewAssetIdentifierStore(tx) - if err := txIdentifierStore.Register(ctx, card.ICCID, model.AssetTypeIotCard, iotCard.ID); err != nil { - return fmt.Errorf("ICCID 已被占用: %s", card.ICCID) - } - if card.VirtualNo != "" { - if err := txIdentifierStore.Register(ctx, card.VirtualNo, model.AssetTypeIotCard, iotCard.ID); err != nil { - return fmt.Errorf("虚拟号已被占用: %s", card.VirtualNo) - } - } - return nil - }) - if txErr != nil { - h.logger.Error("创建 IoT 卡失败", - zap.String("iccid", card.ICCID), - zap.Error(txErr), - ) - result.failedItems = append(result.failedItems, model.ImportResultItem{ - Line: meta.line, - ICCID: card.ICCID, - MSISDN: meta.msisdn, - Reason: txErr.Error(), - }) - result.failCount++ - continue - } - createdCards = append(createdCards, iotCard) - result.successCount++ + iotCards = append(iotCards, iotCard) + pending = append(pending, pendingEntry{item: card, meta: meta}) } - h.batchCreateWallets(ctx, createdCards) + if len(iotCards) == 0 { + return + } + + // 整批一个事务：批量写入 iot_card + asset_identifier + // 相比逐卡事务，将 N 个事务压缩为 1 个，大幅减少数据库往返 + txErr := h.db.Transaction(func(tx *gorm.DB) error { + // 批量插入 iot_card，GORM 会通过 RETURNING 回填所有 ID + if err := tx.CreateInBatches(&iotCards, 500).Error; err != nil { + return err + } + // 用回填的 ID 构建 asset_identifier 列表 + identifiers := make([]*model.AssetIdentifier, 0, len(iotCards)*2) + for _, c := range iotCards { + identifiers = append(identifiers, &model.AssetIdentifier{ + Identifier: c.ICCID, + AssetType: model.AssetTypeIotCard, + AssetID: c.ID, + }) + if c.VirtualNo != "" { + identifiers = append(identifiers, &model.AssetIdentifier{ + Identifier: c.VirtualNo, + AssetType: model.AssetTypeIotCard, + AssetID: c.ID, + }) + } + } + return tx.CreateInBatches(&identifiers, 500).Error + }) + + if txErr != nil { + h.logger.Error("批量创建 IoT 卡失败", + zap.Error(txErr), + zap.Int("batch_size", len(iotCards)), + ) + for _, p := range pending { + result.failedItems = append(result.failedItems, model.ImportResultItem{ + Line: p.meta.line, + ICCID: p.item.ICCID, + MSISDN: p.meta.msisdn, + Reason: "批量写入失败: " + txErr.Error(), + }) + result.failCount++ + } + return + } + + result.successCount += len(iotCards) + + h.batchCreateWallets(ctx, iotCards) if h.pollingCallback != nil { - h.pollingCallback.OnBatchCardsCreated(ctx, createdCards) + h.pollingCallback.OnBatchCardsCreated(ctx, iotCards) } } diff --git a/pkg/queue/handler.go b/pkg/queue/handler.go index 2844571..1091c9f 100644 --- a/pkg/queue/handler.go +++ b/pkg/queue/handler.go @@ -88,7 +88,6 @@ func (h *Handler) registerIotCardImportHandler() { h.workerResult.Stores.IotCardImportTask, h.workerResult.Stores.IotCard, h.workerResult.Stores.AssetWallet, - h.workerResult.Stores.AssetIdentifier, h.storage, h.pollingCallback, h.logger,