package task import ( "context" "encoding/csv" "io" "os" "strings" "github.com/bytedance/sonic" "github.com/hibiken/asynq" "go.uber.org/zap" "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/internal/store/postgres" "github.com/break/junhong_cmp_fiber/pkg/constants" "github.com/break/junhong_cmp_fiber/pkg/storage" ) // OrderPackageInvalidatePayload 批量失效订单套餐任务载荷 type OrderPackageInvalidatePayload struct { TaskID uint `json:"task_id"` } // OrderPackageInvalidateHandler 批量失效订单套餐任务处理器 type OrderPackageInvalidateHandler struct { taskStore *postgres.OrderPackageInvalidateTaskStore orderStore *postgres.OrderStore packageUsageStore *postgres.PackageUsageStore storageService *storage.Service logger *zap.Logger } // NewOrderPackageInvalidateHandler 创建处理器实例 func NewOrderPackageInvalidateHandler( taskStore *postgres.OrderPackageInvalidateTaskStore, orderStore *postgres.OrderStore, packageUsageStore *postgres.PackageUsageStore, storageSvc *storage.Service, logger *zap.Logger, ) *OrderPackageInvalidateHandler { return &OrderPackageInvalidateHandler{ taskStore: taskStore, orderStore: orderStore, packageUsageStore: packageUsageStore, storageService: storageSvc, logger: logger, } } // Handle 处理批量失效订单套餐任务 // POST /api/admin/order-package-invalidate-tasks func (h *OrderPackageInvalidateHandler) Handle(ctx context.Context, t *asynq.Task) error { var payload OrderPackageInvalidatePayload if err := sonic.Unmarshal(t.Payload(), &payload); err != nil { h.logger.Error("解析批量失效任务载荷失败", zap.Error(err), zap.String("task_id", t.ResultWriter().TaskID()), ) return asynq.SkipRetry } importTask, err := h.taskStore.GetByID(ctx, payload.TaskID) if err != nil { h.logger.Error("获取批量失效任务失败", zap.Uint("task_id", payload.TaskID), zap.Error(err), ) return asynq.SkipRetry } if importTask.Status != model.ImportTaskStatusPending { h.logger.Info("批量失效任务已处理,跳过", zap.Uint("task_id", payload.TaskID), zap.Int("status", importTask.Status), ) return nil } h.taskStore.UpdateStatus(ctx, importTask.ID, model.ImportTaskStatusProcessing, "") h.logger.Info("开始处理批量失效订单套餐任务", zap.Uint("task_id", importTask.ID), zap.String("task_no", importTask.TaskNo), ) orderNos, err := h.downloadAndParseCSV(ctx, importTask) if err != nil { h.logger.Error("下载或解析 CSV 失败", zap.Uint("task_id", importTask.ID), zap.Error(err), ) h.taskStore.UpdateStatus(ctx, importTask.ID, model.ImportTaskStatusFailed, err.Error()) return asynq.SkipRetry } successCount, failedItems := h.processRows(ctx, orderNos) failCount := len(failedItems) totalCount := len(orderNos) h.taskStore.UpdateResult(ctx, importTask.ID, totalCount, successCount, failCount, model.ImportResultItems(toImportResultItems(failedItems))) if failCount > 0 && successCount == 0 { h.taskStore.UpdateStatus(ctx, importTask.ID, model.ImportTaskStatusFailed, "所有行均处理失败") } else { h.taskStore.UpdateStatus(ctx, importTask.ID, model.ImportTaskStatusCompleted, "") } h.logger.Info("批量失效订单套餐任务完成", zap.Uint("task_id", importTask.ID), zap.Int("total", totalCount), zap.Int("success", successCount), zap.Int("fail", failCount), ) return nil } // invalidateRow 单行处理结果 type invalidateRow struct { line int orderNo string reason string } // processRows 逐行处理订单号,返回成功数和失败列表 func (h *OrderPackageInvalidateHandler) processRows(ctx context.Context, rows []string) (int, []invalidateRow) { successCount := 0 var failed []invalidateRow for i, orderNo := range rows { line := i + 2 // 第1行为表头,数据从第2行开始 if err := h.processOneOrder(ctx, orderNo); err != nil { failed = append(failed, invalidateRow{line: line, orderNo: orderNo, reason: err.Error()}) } else { successCount++ } } return successCount, failed } // processOneOrder 处理单个订单号:查订单 → 查套餐 → 批量更新状态=4 func (h *OrderPackageInvalidateHandler) processOneOrder(ctx context.Context, orderNo string) error { order, err := h.orderStore.GetByOrderNo(ctx, orderNo) if err != nil { return errOrderNotFound(orderNo) } usages, err := h.packageUsageStore.ListActiveByOrderID(ctx, order.ID) if err != nil { return errQueryFailed(orderNo) } if len(usages) == 0 { // 套餐全部已是终态,视为成功 return nil } ids := make([]uint, 0, len(usages)) for _, u := range usages { ids = append(ids, u.ID) } if err := h.packageUsageStore.BatchUpdateStatus(ctx, ids, constants.PackageUsageStatusInvalidated); err != nil { return errUpdateFailed(orderNo) } return nil } // downloadAndParseCSV 从对象存储下载 CSV 并解析 order_no 列 func (h *OrderPackageInvalidateHandler) downloadAndParseCSV(ctx context.Context, task *model.OrderPackageInvalidateTask) ([]string, error) { if h.storageService == nil { return nil, ErrStorageNotConfigured } if task.StorageKey == "" { return nil, ErrStorageKeyEmpty } localPath, cleanup, err := h.storageService.DownloadToTemp(ctx, task.StorageKey) if err != nil { return nil, err } defer cleanup() return parseOrderNoCSV(localPath) } // parseOrderNoCSV 解析单列 CSV(第一列为 order_no,首行为表头) func parseOrderNoCSV(filePath string) ([]string, error) { f, err := os.Open(filePath) if err != nil { return nil, err } defer f.Close() reader := csv.NewReader(f) reader.TrimLeadingSpace = true var orderNos []string firstRow := true for { record, err := reader.Read() if err == io.EOF { break } if err != nil { return nil, err } if firstRow { firstRow = false continue // 跳过表头 } if len(record) == 0 { continue } orderNo := strings.TrimSpace(record[0]) if orderNo == "" { continue } orderNos = append(orderNos, orderNo) } return orderNos, nil } // toImportResultItems 将内部失败记录转换为通用失败明细格式 // 复用 ImportResultItem,ICCID 字段存储 order_no func toImportResultItems(rows []invalidateRow) []model.ImportResultItem { items := make([]model.ImportResultItem, 0, len(rows)) for _, r := range rows { items = append(items, model.ImportResultItem{ Line: r.line, ICCID: r.orderNo, Reason: r.reason, }) } return items } func errOrderNotFound(orderNo string) error { return orderInvalidateError("订单不存在: " + orderNo) } func errQueryFailed(orderNo string) error { return orderInvalidateError("查询套餐失败: " + orderNo) } func errUpdateFailed(orderNo string) error { return orderInvalidateError("更新套餐状态失败: " + orderNo) } type orderInvalidateError string func (e orderInvalidateError) Error() string { return string(e) }