junhong_cmp_fiber/internal/task/order_package_invalidate.go

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