junhong_cmp_fiber/internal/service/iot_card_import/service.go

package iot_card_import

import (
	"context"
	"path/filepath"
	"strconv"
	"time"

	"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
	"github.com/break/junhong_cmp_fiber/internal/model"
	"github.com/break/junhong_cmp_fiber/internal/model/dto"
	"github.com/break/junhong_cmp_fiber/internal/store"
	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/errors"
	"github.com/break/junhong_cmp_fiber/pkg/middleware"
	"github.com/break/junhong_cmp_fiber/pkg/queue"
	"github.com/hibiken/asynq"
	"gorm.io/gorm"
)

type Service struct {
	db              *gorm.DB
	importTaskStore *postgres.IotCardImportTaskStore
	carrierStore    carrierGetter
	queueClient     *queue.Client
	auditWriter     *audit.Writer
}

type carrierGetter interface {
	GetByID(ctx context.Context, id uint) (*model.Carrier, error)
}

type CarrierStore struct {
	db *gorm.DB
}

func NewCarrierStore(db *gorm.DB) *CarrierStore {
	return &CarrierStore{db: db}
}

func (s *CarrierStore) GetByID(ctx context.Context, id uint) (*model.Carrier, error) {
	var carrier model.Carrier
	if err := s.db.WithContext(ctx).First(&carrier, id).Error; err != nil {
		return nil, err
	}
	return &carrier, nil
}

func New(
	db *gorm.DB,
	importTaskStore *postgres.IotCardImportTaskStore,
	queueClient *queue.Client,
	auditWriters ...*audit.Writer,
) *Service {
	service := &Service{
		db:              db,
		importTaskStore: importTaskStore,
		carrierStore:    NewCarrierStore(db),
		queueClient:     queueClient,
	}
	if len(auditWriters) > 0 {
		service.auditWriter = auditWriters[0]
	}
	return service
}

type IotCardImportPayload struct {
	TaskID uint `json:"task_id"`
}

func (s *Service) CreateImportTask(ctx context.Context, req *dto.ImportIotCardRequest) (*dto.ImportIotCardResponse, error) {
	userID := middleware.GetUserIDFromContext(ctx)
	if userID == 0 {
		return nil, errors.New(errors.CodeUnauthorized, "未授权访问")
	}

	carrier, err := s.carrierStore.GetByID(ctx, req.CarrierID)
	if err != nil {
		return nil, errors.New(errors.CodeInvalidParam, "运营商不存在")
	}

	taskNo := s.importTaskStore.GenerateTaskNo(ctx)
	fileName := filepath.Base(req.FileKey)

	cardCategory := req.CardCategory
	if cardCategory == "" {
		cardCategory = constants.CardCategoryNormal
	}

	task := &model.IotCardImportTask{
		TaskNo:         taskNo,
		Status:         model.ImportTaskStatusPending,
		CarrierID:      req.CarrierID,
		CarrierType:    carrier.CarrierType,
		CarrierName:    carrier.CarrierName,
		BatchNo:        req.BatchNo,
		FileName:       fileName,
		StorageKey:     req.FileKey,
		CardCategory:   cardCategory,
		RealnamePolicy: req.RealnamePolicy,
		CreatorName:    middleware.GetUsernameFromContext(ctx),
	}
	task.Creator = userID
	task.Updater = userID

	if s.auditWriter == nil {
		return nil, errors.New(errors.CodeInvalidStatus, "IoT 卡导入任务统一审计接缝未配置")
	}
	if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := tx.WithContext(ctx).Create(task).Error; err != nil {
			return err
		}
		return s.writeImportTaskAudit(ctx, tx, task, nil, importTaskState(task), constants.AuditResultSuccess, "created", "", "")
	}); err != nil {
		appErr := errors.Wrap(errors.CodeInternalError, err, "创建导入任务失败")
		s.recordImportTaskAudit(ctx, task, nil, importTaskState(task), constants.AuditResultFailed, "create_failed", errors.CodeDatabaseError, "创建 IoT 卡导入任务失败")
		return nil, appErr
	}

	payload := IotCardImportPayload{TaskID: task.ID}
	err = s.queueClient.EnqueueTask(
		ctx,
		constants.TaskTypeIotCardImport,
		payload,
		asynq.Queue(constants.QueueForTaskType(constants.TaskTypeIotCardImport)),
	)
	if err != nil {
		secondaryErr := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
			before := importTaskState(task)
			if updateErr := tx.WithContext(ctx).Model(&model.IotCardImportTask{}).Where("id = ?", task.ID).Updates(map[string]any{
				"status": model.ImportTaskStatusFailed, "error_message": "任务入队失败", "completed_at": time.Now(), "updated_at": time.Now(),
			}).Error; updateErr != nil {
				return updateErr
			}
			task.Status, task.ErrorMessage = model.ImportTaskStatusFailed, "任务入队失败"
			return s.writeImportTaskAudit(ctx, tx, task, before, importTaskState(task), constants.AuditResultFailed, "enqueue_failed", strconv.Itoa(errors.CodeTaskQueueError), "IoT 卡导入任务入队失败")
		})
		if secondaryErr != nil {
			s.recordImportTaskAudit(ctx, task, nil, importTaskState(task), constants.AuditResultFailed, "enqueue_audit_failed", errors.CodeTaskQueueError, "IoT 卡导入任务入队失败")
		}
		appErr := errors.Wrap(errors.CodeInternalError, err, "任务入队失败")
		return nil, appErr
	}

	return &dto.ImportIotCardResponse{
		TaskID:  task.ID,
		TaskNo:  taskNo,
		Message: "导入任务已创建，Worker 将异步处理文件",
	}, nil
}

func (s *Service) List(ctx context.Context, req *dto.ListImportTaskRequest, startTime, endTime *time.Time) (*dto.ListImportTaskResponse, error) {
	page := req.Page
	pageSize := req.PageSize
	if page == 0 {
		page = 1
	}
	if pageSize == 0 {
		pageSize = constants.DefaultPageSize
	}

	opts := &store.QueryOptions{
		Page:     page,
		PageSize: pageSize,
	}

	filters := make(map[string]interface{})
	if req.Status != nil {
		filters["status"] = *req.Status
	}
	if req.CarrierID != nil {
		filters["carrier_id"] = *req.CarrierID
	}
	if req.BatchNo != "" {
		filters["batch_no"] = req.BatchNo
	}
	if startTime != nil {
		filters["start_time"] = *startTime
	}
	if endTime != nil {
		filters["end_time"] = *endTime
	}

	tasks, total, err := s.importTaskStore.List(ctx, opts, filters)
	if err != nil {
		return nil, err
	}

	list := make([]*dto.ImportTaskResponse, 0, len(tasks))
	for _, task := range tasks {
		list = append(list, s.toTaskResponse(task))
	}

	return &dto.ListImportTaskResponse{
		List:     list,
		Total:    total,
		Page:     page,
		PageSize: pageSize,
	}, nil
}

func (s *Service) GetByID(ctx context.Context, id uint) (*dto.ImportTaskDetailResponse, error) {
	task, err := s.importTaskStore.GetByID(ctx, id)
	if err != nil {
		return nil, errors.New(errors.CodeNotFound, "导入任务不存在")
	}

	resp := &dto.ImportTaskDetailResponse{
		ImportTaskResponse: *s.toTaskResponse(task),
		SkippedItems:       make([]*dto.ImportResultItemDTO, 0),
		FailedItems:        make([]*dto.ImportResultItemDTO, 0),
	}

	for _, item := range task.SkippedItems {
		resp.SkippedItems = append(resp.SkippedItems, &dto.ImportResultItemDTO{
			Line:   item.Line,
			ICCID:  item.ICCID,
			MSISDN: item.MSISDN,
			Reason: item.Reason,
		})
	}

	for _, item := range task.FailedItems {
		resp.FailedItems = append(resp.FailedItems, &dto.ImportResultItemDTO{
			Line:   item.Line,
			ICCID:  item.ICCID,
			MSISDN: item.MSISDN,
			Reason: item.Reason,
		})
	}

	return resp, nil
}

func (s *Service) toTaskResponse(task *model.IotCardImportTask) *dto.ImportTaskResponse {
	var startedAt, completedAt *time.Time
	if task.StartedAt != nil {
		startedAt = task.StartedAt
	}
	if task.CompletedAt != nil {
		completedAt = task.CompletedAt
	}

	return &dto.ImportTaskResponse{
		ID:             task.ID,
		TaskNo:         task.TaskNo,
		Status:         task.Status,
		StatusText:     getStatusText(task.Status),
		CarrierID:      task.CarrierID,
		CarrierType:    task.CarrierType,
		CarrierName:    task.CarrierName,
		BatchNo:        task.BatchNo,
		CardCategory:   task.CardCategory,
		RealnamePolicy: task.RealnamePolicy,
		FileName:       task.FileName,
		TotalCount:     task.TotalCount,
		SuccessCount:   task.SuccessCount,
		SkipCount:      task.SkipCount,
		FailCount:      task.FailCount,
		StartedAt:      startedAt,
		CompletedAt:    completedAt,
		ErrorMessage:   task.ErrorMessage,
		CreatorName:    task.CreatorName,
		CreatedAt:      task.CreatedAt,
	}
}

func getStatusText(status int) string {
	switch status {
	case model.ImportTaskStatusPending:
		return "待处理"
	case model.ImportTaskStatusProcessing:
		return "处理中"
	case model.ImportTaskStatusCompleted:
		return "已完成"
	case model.ImportTaskStatusFailed:
		return "失败"
	default:
		return "未知"
	}
}