break
5e552d99bc
Constraint: 在线热修前必须保存当前迭代分支全部有效代码进展 Confidence: medium Scope-risk: broad Directive: 后续修改需保持审计事件与业务事务边界一致 Tested: git diff --cached --check Not-tested: 未运行全量测试，提交用于切换分支前保存既有工作
370 lines
13 KiB
Go
370 lines
13 KiB
Go
package task
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/csv"
|
||
"fmt"
|
||
"io"
|
||
"os"
|
||
"strings"
|
||
|
||
"github.com/bytedance/sonic"
|
||
"github.com/hibiken/asynq"
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||
"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
|
||
"github.com/break/junhong_cmp_fiber/pkg/auditfailure"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
pkgerrors "github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
"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
|
||
auditWriter *audit.Writer
|
||
}
|
||
|
||
// NewOrderPackageInvalidateHandler 创建处理器实例
|
||
func NewOrderPackageInvalidateHandler(
|
||
taskStore *postgres.OrderPackageInvalidateTaskStore,
|
||
orderStore *postgres.OrderStore,
|
||
packageUsageStore *postgres.PackageUsageStore,
|
||
storageSvc *storage.Service,
|
||
logger *zap.Logger,
|
||
auditWriters ...*audit.Writer,
|
||
) *OrderPackageInvalidateHandler {
|
||
handler := &OrderPackageInvalidateHandler{
|
||
taskStore: taskStore,
|
||
orderStore: orderStore,
|
||
packageUsageStore: packageUsageStore,
|
||
storageService: storageSvc,
|
||
logger: logger,
|
||
}
|
||
if len(auditWriters) > 0 {
|
||
handler.auditWriter = auditWriters[0]
|
||
}
|
||
return handler
|
||
}
|
||
|
||
// 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
|
||
}
|
||
rootEventID := audit.TaskEventID(constants.AuditResourceOrderPackageInvalidateTask, importTask.ID, "completed")
|
||
ctx = auditcontext.With(ctx, auditcontext.Context{
|
||
ActorKind: constants.AuditActorSystemTask, ActorID: constants.TaskTypeOrderPackageInvalidate,
|
||
ActorName: "订单套餐批量失效任务", Source: constants.AuditSourceWorker,
|
||
CorrelationID: importTask.TaskNo, ParentEventID: rootEventID,
|
||
})
|
||
claimed, err := h.taskStore.Claim(ctx, importTask.ID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !claimed {
|
||
h.logger.Info("批量失效任务已处理，跳过",
|
||
zap.Uint("task_id", payload.TaskID),
|
||
zap.Int("status", importTask.Status),
|
||
)
|
||
return nil
|
||
}
|
||
|
||
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),
|
||
)
|
||
if finishErr := h.finishInvalidateTask(ctx, importTask, 0, 0, 1, model.ImportTaskStatusFailed, err.Error(), nil); finishErr != nil {
|
||
h.resetInvalidateTaskForRetry(ctx, importTask.ID)
|
||
return finishErr
|
||
}
|
||
return asynq.SkipRetry
|
||
}
|
||
|
||
successCount, failedItems := h.processRows(ctx, importTask.ID, orderNos)
|
||
failCount := len(failedItems)
|
||
totalCount := len(orderNos)
|
||
|
||
status, errorMessage := model.ImportTaskStatusCompleted, ""
|
||
if failCount > 0 && successCount == 0 {
|
||
status, errorMessage = model.ImportTaskStatusFailed, "所有行均处理失败"
|
||
}
|
||
if err := h.finishInvalidateTask(ctx, importTask, totalCount, successCount, failCount, status, errorMessage, model.ImportResultItems(toImportResultItems(failedItems))); err != nil {
|
||
h.resetInvalidateTaskForRetry(ctx, importTask.ID)
|
||
return err
|
||
}
|
||
|
||
h.logger.Info("批量失效订单套餐任务完成",
|
||
zap.Uint("task_id", importTask.ID),
|
||
zap.Int("total", totalCount),
|
||
zap.Int("success", successCount),
|
||
zap.Int("fail", failCount),
|
||
)
|
||
|
||
return nil
|
||
}
|
||
|
||
func (h *OrderPackageInvalidateHandler) resetInvalidateTaskForRetry(ctx context.Context, taskID uint) {
|
||
_ = h.taskStore.DB().WithContext(ctx).Model(&model.OrderPackageInvalidateTask{}).
|
||
Where("id = ? AND status = ?", taskID, model.ImportTaskStatusProcessing).
|
||
Updates(map[string]any{"status": model.ImportTaskStatusPending, "started_at": nil}).Error
|
||
}
|
||
|
||
// invalidateRow 单行处理结果
|
||
type invalidateRow struct {
|
||
line int
|
||
orderNo string
|
||
reason string
|
||
}
|
||
|
||
// processRows 逐行处理订单号，返回成功数和失败列表
|
||
func (h *OrderPackageInvalidateHandler) processRows(ctx context.Context, taskID uint, rows []string) (int, []invalidateRow) {
|
||
successCount := 0
|
||
var failed []invalidateRow
|
||
|
||
for i, orderNo := range rows {
|
||
line := i + 2 // 第1行为表头，数据从第2行开始
|
||
if err := h.processOneOrder(ctx, taskID, 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, taskID uint, orderNo string) error {
|
||
order, err := h.orderStore.GetByOrderNo(ctx, orderNo)
|
||
if err != nil {
|
||
h.appendInvalidateFailure(ctx, taskID, &model.Order{OrderNo: orderNo}, "订单不存在")
|
||
return errOrderNotFound(orderNo)
|
||
}
|
||
|
||
queryFailed := false
|
||
err = h.taskStore.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
usages, queryErr := postgres.NewPackageUsageStore(tx, nil).ListActiveByOrderID(ctx, order.ID)
|
||
if queryErr != nil {
|
||
queryFailed = true
|
||
return queryErr
|
||
}
|
||
if len(usages) == 0 {
|
||
return nil
|
||
}
|
||
ids := make([]uint, 0, len(usages))
|
||
resources := []audit.ResourceInput{audit.OrderResource(order, constants.AuditResourceRelationPrimary, constants.AuditResourceRoleOrderTarget)}
|
||
for _, usage := range usages {
|
||
ids = append(ids, usage.ID)
|
||
resources = append(resources, audit.PackageUsageResource(usage, constants.AuditResourceRelationAffected, constants.AuditResourceRolePackageUsageTarget,
|
||
map[string]any{"status": usage.Status}, map[string]any{"status": constants.PackageUsageStatusInvalidated}))
|
||
}
|
||
if err := postgres.NewPackageUsageStore(tx, nil).BatchUpdateStatus(ctx, ids, constants.PackageUsageStatusInvalidated); err != nil {
|
||
return err
|
||
}
|
||
return h.auditWriter.Append(ctx, tx, audit.AppendInput{
|
||
EventID: audit.TaskEventID(constants.AuditResourceOrderPackageInvalidateTask, taskID, fmt.Sprintf("item:%d", order.ID)),
|
||
ActionCode: constants.AuditActionOrderPackageInvalidateItem, Summary: "失效订单套餐权益",
|
||
ScopeType: constants.AuditScopePlatform, Result: constants.AuditResultSuccess, Resources: resources,
|
||
})
|
||
})
|
||
if err != nil {
|
||
summary := "更新套餐状态失败"
|
||
if queryFailed {
|
||
summary = "查询套餐失败"
|
||
}
|
||
h.appendInvalidateFailure(ctx, taskID, order, summary)
|
||
if queryFailed {
|
||
return errQueryFailed(orderNo)
|
||
}
|
||
return errUpdateFailed(orderNo)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (h *OrderPackageInvalidateHandler) appendInvalidateFailure(ctx context.Context, taskID uint, order *model.Order, summary string) {
|
||
if h.auditWriter == nil || order == nil || order.OrderNo == "" {
|
||
return
|
||
}
|
||
err := h.taskStore.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
keyHash := sha256.Sum256([]byte(order.OrderNo))
|
||
return h.auditWriter.Append(ctx, tx, audit.AppendInput{
|
||
EventID: fmt.Sprintf("task:order_invalidate:%d:failed:%x", taskID, keyHash[:6]),
|
||
ActionCode: constants.AuditActionOrderPackageInvalidateItem, Summary: summary,
|
||
ScopeType: constants.AuditScopePlatform, Result: constants.AuditResultFailed,
|
||
ErrorCode: fmt.Sprintf("%d", pkgerrors.CodeDatabaseError), ErrorSummary: summary,
|
||
Resources: []audit.ResourceInput{audit.OrderResource(order, constants.AuditResourceRelationPrimary, constants.AuditResourceRoleOrderTarget)},
|
||
})
|
||
})
|
||
if err != nil {
|
||
auditfailure.RecordSecondaryWriteFailure(constants.AuditActionOrderPackageInvalidateItem, order.OrderNo, "", auditcontext.From(ctx).CorrelationID, fmt.Sprintf("%d", pkgerrors.CodeDatabaseError), err)
|
||
}
|
||
}
|
||
|
||
func (h *OrderPackageInvalidateHandler) finishInvalidateTask(ctx context.Context, task *model.OrderPackageInvalidateTask, totalCount, successCount, failCount, status int, errorMessage string, failedItems model.ImportResultItems) error {
|
||
if h.auditWriter == nil {
|
||
return pkgerrors.New(pkgerrors.CodeInvalidStatus, "订单套餐失效任务统一审计接缝未配置")
|
||
}
|
||
return h.taskStore.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
txStore := h.taskStore.WithTx(tx)
|
||
if err := txStore.UpdateResult(ctx, task.ID, totalCount, successCount, failCount, failedItems); err != nil {
|
||
return err
|
||
}
|
||
if err := txStore.UpdateStatus(ctx, task.ID, status, errorMessage); err != nil {
|
||
return err
|
||
}
|
||
rootID := audit.TaskEventID(constants.AuditResourceOrderPackageInvalidateTask, task.ID, "completed")
|
||
var actualSuccess, actualFail int64
|
||
if err := tx.WithContext(ctx).Model(&model.AuditEvent{}).
|
||
Where("parent_event_id = ? AND action_code = ? AND result = ?", rootID, constants.AuditActionOrderPackageInvalidateItem, constants.AuditResultSuccess).
|
||
Count(&actualSuccess).Error; err != nil {
|
||
return err
|
||
}
|
||
if err := tx.WithContext(ctx).Model(&model.AuditEvent{}).
|
||
Where("parent_event_id = ? AND action_code = ? AND result = ?", rootID, constants.AuditActionOrderPackageInvalidateItem, constants.AuditResultFailed).
|
||
Count(&actualFail).Error; err != nil {
|
||
return err
|
||
}
|
||
auditFailCount := int(actualFail)
|
||
if auditFailCount == 0 && failCount > 0 {
|
||
auditFailCount = failCount
|
||
}
|
||
return h.auditWriter.WriteTask(ctx, tx, audit.TaskInput{
|
||
EventID: rootID, ActionCode: constants.AuditActionOrderPackageInvalidateTaskCompleted,
|
||
Summary: "完成订单套餐批量失效任务", TaskID: task.ID, TaskNo: task.TaskNo,
|
||
Result: batchAuditResult(int(actualSuccess), auditFailCount), CorrelationID: task.TaskNo,
|
||
ParentEventID: audit.TaskEventID(constants.AuditResourceOrderPackageInvalidateTask, task.ID, "created"),
|
||
BatchTotal: int(actualSuccess) + auditFailCount, SuccessCount: int(actualSuccess), FailCount: auditFailCount,
|
||
IdentitySnapshot: map[string]any{"id": task.ID, "task_no": task.TaskNo, "file_name": task.FileName},
|
||
BeforeData: map[string]any{"status": model.ImportTaskStatusProcessing},
|
||
AfterData: map[string]any{
|
||
"status": status, "total_count": totalCount, "success_count": successCount, "fail_count": failCount,
|
||
},
|
||
Metadata: map[string]any{"task_success_count": successCount, "task_fail_count": failCount},
|
||
})
|
||
})
|
||
}
|
||
|
||
// 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) }
|