收口审计治理与套餐任务进展

Constraint: 在线热修前必须保存当前迭代分支全部有效代码进展 Confidence: medium Scope-risk: broad Directive: 后续修改需保持审计事件与业务事务边界一致 Tested: git diff --cached --check Not-tested: 未运行全量测试，提交用于切换分支前保存既有工作
2026-08-05 14:30:54 +08:00
parent b3499adfca
commit 5e552d99bc
178 changed files with 16797 additions and 5674 deletions
--- a/internal/task/order_package_invalidate.go
+++ b/internal/task/order_package_invalidate.go
@@ -2,7 +2,9 @@ package task

 import (
 	"context"
+	"crypto/sha256"
 	"encoding/csv"
+	"fmt"
 	"io"
 	"os"
 	"strings"
@@ -10,10 +12,15 @@ import (
 	"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"
 )

@@ -29,6 +36,7 @@ type OrderPackageInvalidateHandler struct {
 	packageUsageStore *postgres.PackageUsageStore
 	storageService    *storage.Service
 	logger            *zap.Logger
+	auditWriter       *audit.Writer
 }

 // NewOrderPackageInvalidateHandler 创建处理器实例
@@ -38,14 +46,19 @@ func NewOrderPackageInvalidateHandler(
 	packageUsageStore *postgres.PackageUsageStore,
 	storageSvc *storage.Service,
 	logger *zap.Logger,
+	auditWriters ...*audit.Writer,
 ) *OrderPackageInvalidateHandler {
-	return &OrderPackageInvalidateHandler{
+	handler := &OrderPackageInvalidateHandler{
 		taskStore:         taskStore,
 		orderStore:        orderStore,
 		packageUsageStore: packageUsageStore,
 		storageService:    storageSvc,
 		logger:            logger,
 	}
+	if len(auditWriters) > 0 {
+		handler.auditWriter = auditWriters[0]
+	}
+	return handler
 }

 // Handle 处理批量失效订单套餐任务
@@ -68,8 +81,17 @@ func (h *OrderPackageInvalidateHandler) Handle(ctx context.Context, t *asynq.Tas
 		)
 		return asynq.SkipRetry
 	}
-
-	if importTask.Status != model.ImportTaskStatusPending {
+	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),
@@ -77,8 +99,6 @@ func (h *OrderPackageInvalidateHandler) Handle(ctx context.Context, t *asynq.Tas
 		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),
@@ -90,21 +110,24 @@ func (h *OrderPackageInvalidateHandler) Handle(ctx context.Context, t *asynq.Tas
 			zap.Uint("task_id", importTask.ID),
 			zap.Error(err),
 		)
-		h.taskStore.UpdateStatus(ctx, importTask.ID, model.ImportTaskStatusFailed, err.Error())
+		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, orderNos)
+	successCount, failedItems := h.processRows(ctx, importTask.ID, orderNos)
 	failCount := len(failedItems)
 	totalCount := len(orderNos)

-	h.taskStore.UpdateResult(ctx, importTask.ID, totalCount, successCount, failCount,
-		model.ImportResultItems(toImportResultItems(failedItems)))
-
+	status, errorMessage := model.ImportTaskStatusCompleted, ""
 	if failCount > 0 && successCount == 0 {
-		h.taskStore.UpdateStatus(ctx, importTask.ID, model.ImportTaskStatusFailed, "所有行均处理失败")
-	} else {
-		h.taskStore.UpdateStatus(ctx, importTask.ID, model.ImportTaskStatusCompleted, "")
+		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("批量失效订单套餐任务完成",
@@ -117,6 +140,12 @@ func (h *OrderPackageInvalidateHandler) Handle(ctx context.Context, t *asynq.Tas
 	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
@@ -125,13 +154,13 @@ type invalidateRow struct {
 }

 // processRows 逐行处理订单号，返回成功数和失败列表
-func (h *OrderPackageInvalidateHandler) processRows(ctx context.Context, rows []string) (int, []invalidateRow) {
+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, orderNo); err != nil {
+		if err := h.processOneOrder(ctx, taskID, orderNo); err != nil {
 			failed = append(failed, invalidateRow{line: line, orderNo: orderNo, reason: err.Error()})
 		} else {
 			successCount++
@@ -142,34 +171,117 @@ func (h *OrderPackageInvalidateHandler) processRows(ctx context.Context, rows []
 }

 // processOneOrder 处理单个订单号：查订单 → 查套餐 → 批量更新状态=4
-func (h *OrderPackageInvalidateHandler) processOneOrder(ctx context.Context, orderNo string) error {
+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)
 	}

-	usages, err := h.packageUsageStore.ListActiveByOrderID(ctx, order.ID)
+	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 {
-		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 {
+		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 {