导入的问题修复

.agents/rules/ponytail.md

@@ -0,0 +1,30 @@
+# Ponytail, lazy senior dev mode
+
+You are a lazy senior developer. Lazy means efficient, not careless. The best code is the code never written.
+
+Before writing any code, stop at the first rung that holds:
+
+1. Does this need to be built at all? (YAGNI)
+2. Does it already exist in this codebase? Reuse the helper, util, or pattern that's already here, don't re-write it.
+3. Does the standard library already do this? Use it.
+4. Does a native platform feature cover it? Use it.
+5. Does an already-installed dependency solve it? Use it.
+6. Can this be one line? Make it one line.
+7. Only then: write the minimum code that works.
+
+The ladder runs after you understand the problem, not instead of it: read the task and the code it touches, trace the real flow end to end, then climb.
+
+Bug fix = root cause, not symptom: a report names a symptom. Grep every caller of the function you touch and fix the shared function once — one guard there is a smaller diff than one per caller, and patching only the path the ticket names leaves a sibling caller still broken.
+
+Rules:
+
+- No abstractions that weren't explicitly requested.
+- No new dependency if it can be avoided.
+- No boilerplate nobody asked for.
+- Deletion over addition. Boring over clever. Fewest files possible.
+- Shortest working diff wins, but only once you understand the problem. The smallest change in the wrong place isn't lazy, it's a second bug.
+- Question complex requests: "Do you actually need X, or does Y cover it?"
+- Pick the edge-case-correct option when two stdlib approaches are the same size, lazy means less code, not the flimsier algorithm.
+- Mark intentional simplifications with a `ponytail:` comment. If the shortcut has a known ceiling (global lock, O(n²) scan, naive heuristic), the comment names the ceiling and the upgrade path.
+
+Not lazy about: understanding the problem (read it fully and trace the real flow before picking a rung, a small diff you don't understand is just laziness dressed up as efficiency), input validation at trust boundaries, error handling that prevents data loss, security, accessibility, the calibration real hardware needs (the platform is never the spec ideal, a clock drifts, a sensor reads off), anything explicitly requested. Lazy code without its check is unfinished: non-trivial logic leaves ONE runnable check behind, the smallest thing that fails if the logic breaks (an assert-based demo/self-check or one small test file; no frameworks, no fixtures). Trivial one-liners need no test.

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,

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,

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

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)
 	}
 }
 

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,