diff --git a/internal/handler/admin/refund.go b/internal/handler/admin/refund.go index 5faa7902..fc0a8973 100644 --- a/internal/handler/admin/refund.go +++ b/internal/handler/admin/refund.go @@ -97,6 +97,20 @@ func (h *RefundHandler) RecoverApproval(c *fiber.Ctx) error { return response.Success(c, result) } +// RetryAssetPostProcessing 人工重试退款资产后处理 +// POST /api/admin/refunds/:id/asset-post-processing/retry +func (h *RefundHandler) RetryAssetPostProcessing(c *fiber.Ctx) error { + id, err := strconv.ParseUint(c.Params("id"), 10, 64) + if err != nil || id == 0 { + return errors.New(errors.CodeInvalidParam, "无效的退款申请ID") + } + result, err := h.service.RetryAssetPostProcessingByAdmin(c.UserContext(), uint(id)) + if err != nil { + return err + } + return response.Success(c, result) +} + // Approve 审批通过退款申请 // POST /api/admin/refunds/:id/approve func (h *RefundHandler) Approve(c *fiber.Ctx) error { diff --git a/internal/infrastructure/commissiondelivery/event.go b/internal/infrastructure/commissiondelivery/event.go index 7cab136c..fcef7197 100644 --- a/internal/infrastructure/commissiondelivery/event.go +++ b/internal/infrastructure/commissiondelivery/event.go @@ -133,14 +133,16 @@ func RecoverRefundPostProcessing(ctx context.Context, db *gorm.DB, repository *o } refundStats := RecoveryStats{} var refunds []model.RefundRequest - if err := db.WithContext(ctx).Where("status = ? AND (commission_deducted = ? OR asset_reset = ?)", model.RefundStatusApproved, false, false).Order("id ASC").Limit(limit).Find(&refunds).Error; err != nil { + // 需人工处理与未到下次重试时间的资产后处理不参与扫描，避免长期失败的退款单占满每轮名额。 + if err := db.WithContext(ctx).Where("status = ? AND (commission_deducted = ? OR (asset_reset = ? AND asset_post_status <> ? AND (asset_post_next_at IS NULL OR asset_post_next_at <= ?)))", + model.RefundStatusApproved, false, false, model.RefundAssetPostManualRequired, time.Now()).Order("id ASC").Limit(limit).Find(&refunds).Error; err != nil { logger.Warn("扫描退款后处理失败", zap.Error(err)) } else { for _, refund := range refunds { if !refund.CommissionDeducted { recoverOne(ctx, db, repository, EventRefundCommissionDeduct, refund.ID, refund.OrderID, &refundStats, logger, true) } - if !refund.AssetReset { + if !refund.AssetReset && assetPostDue(refund, time.Now()) { recoverOne(ctx, db, repository, EventRefundAssetProcess, refund.ID, refund.OrderID, &refundStats, logger, true) } } @@ -148,6 +150,14 @@ func RecoverRefundPostProcessing(ctx context.Context, db *gorm.DB, repository *o return refundStats } +// assetPostDue 判断退款资产后处理是否到了可自动重试的时间：需人工处理的不重投，未到下次重试时间的不重投。 +func assetPostDue(refund model.RefundRequest, now time.Time) bool { + if refund.AssetPostStatus == model.RefundAssetPostManualRequired { + return false + } + return refund.AssetPostNextAt == nil || !refund.AssetPostNextAt.After(now) +} + func recoverOne(ctx context.Context, db *gorm.DB, repository *outbox.Repository, eventType string, aggregateID, orderID uint, stats *RecoveryStats, logger *zap.Logger, retryDelivered bool) { eventID := outboxid.Stable(eventType+":", strconv.FormatUint(uint64(aggregateID), 10)) var event model.OutboxEvent diff --git a/internal/infrastructure/gatewaylimit/stop_resume_record.go b/internal/infrastructure/gatewaylimit/stop_resume_record.go index 626db1f0..dc4f929b 100644 --- a/internal/infrastructure/gatewaylimit/stop_resume_record.go +++ b/internal/infrastructure/gatewaylimit/stop_resume_record.go @@ -26,6 +26,8 @@ type StopResumeRecordInput struct { Attempt int CallErr error Duration time.Duration + // SkipReason 非空表示指令在调用上游前结束（已是目标状态、条件不再满足等），按"跳过"记录，原因写入 ProviderMessage。 + SkipReason string } // StopResumeRecorder 把每次停复机上游调用写入停复机记录表；写入失败只记日志与计数，不影响业务回写。 @@ -54,7 +56,11 @@ func (r *StopResumeRecorder) Record(ctx context.Context, input StopResumeRecordI Attempt: input.Attempt, Result: constants.StopResumeRecordSuccess, DurationMS: int(input.Duration.Milliseconds()), CreatedAt: time.Now(), } - if input.CallErr != nil { + if input.SkipReason != "" { + row.Result = constants.StopResumeRecordSkipped + row.ErrorCategory = "" + row.ProviderMessage = input.SkipReason + } else if input.CallErr != nil { row.Result = constants.StopResumeRecordFailed if gwErr, ok := gateway.AsGatewayError(input.CallErr); ok { row.ErrorCategory = gwErr.Category diff --git a/internal/model/refund.go b/internal/model/refund.go index b21e0e96..df3939b9 100644 --- a/internal/model/refund.go +++ b/internal/model/refund.go @@ -77,6 +77,12 @@ type RefundRequest struct { // 后处理标记 CommissionDeducted bool `gorm:"column:commission_deducted;not null;default:false;comment:佣金是否已回扣" json:"commission_deducted"` AssetReset bool `gorm:"column:asset_reset;not null;default:false;comment:退款后资产处理是否完成" json:"asset_reset"` + + // 资产后处理重试状态（refactor-audit-log-persistence） + AssetPostStatus int `gorm:"column:asset_post_status;type:smallint;not null;default:0;comment:资产后处理状态 0-未开始 1-待重试 2-已完成 3-需人工处理" json:"asset_post_status"` + AssetPostAttempts int `gorm:"column:asset_post_attempts;type:int;not null;default:0;comment:资产后处理已尝试次数" json:"asset_post_attempts"` + AssetPostLastError string `gorm:"column:asset_post_last_error;type:varchar(500);not null;default:'';comment:资产后处理最后一次错误摘要" json:"asset_post_last_error"` + AssetPostNextAt *time.Time `gorm:"column:asset_post_next_at;comment:资产后处理下次自动重试时间" json:"asset_post_next_at,omitempty"` } // TableName 指定表名 @@ -136,3 +142,34 @@ func RefundActiveStatuses() []int { func RefundResubmittableStatuses() []int { return []int{RefundStatusRejected, RefundStatusReturned, RefundStatusChannelFailed, RefundStatusSubmissionFailed} } + +// 退款资产后处理状态。 +const ( + RefundAssetPostNotStarted = 0 // 未开始 + RefundAssetPostRetryPending = 1 // 待重试 + RefundAssetPostCompleted = 2 // 已完成 + RefundAssetPostManualRequired = 3 // 需人工处理 +) + +// RefundPostProcessAttempt 退款资产后处理逐次尝试记录，只新增不修改。 +type RefundPostProcessAttempt struct { + ID uint `gorm:"column:id;primaryKey;autoIncrement" json:"id"` + RefundID uint `gorm:"column:refund_id;not null" json:"refund_id"` + RefundNo string `gorm:"column:refund_no;type:varchar(64);not null;default:''" json:"refund_no"` + Step string `gorm:"column:step;type:varchar(32);not null" json:"step"` + AssetType string `gorm:"column:asset_type;type:varchar(16);not null;default:''" json:"asset_type"` + AssetID uint `gorm:"column:asset_id;not null;default:0" json:"asset_id"` + ICCID string `gorm:"column:iccid;type:varchar(30);not null;default:''" json:"iccid"` + Attempt int `gorm:"column:attempt;not null" json:"attempt"` + Trigger string `gorm:"column:trigger;type:varchar(16);not null" json:"trigger"` + Result string `gorm:"column:result;type:varchar(16);not null" json:"result"` + ErrorCode string `gorm:"column:error_code;type:varchar(32);not null;default:''" json:"error_code"` + ErrorMessage string `gorm:"column:error_message;type:varchar(500);not null;default:''" json:"error_message"` + DurationMS int `gorm:"column:duration_ms;not null;default:0" json:"duration_ms"` + CreatedAt time.Time `gorm:"column:created_at;type:timestamptz;not null;autoCreateTime" json:"created_at"` +} + +// TableName 指定退款资产后处理尝试记录表名。 +func (RefundPostProcessAttempt) TableName() string { + return "tb_refund_post_process_attempt" +} diff --git a/internal/routes/refund.go b/internal/routes/refund.go index a29e8447..336e18f1 100644 --- a/internal/routes/refund.go +++ b/internal/routes/refund.go @@ -74,6 +74,10 @@ func registerRefundRoutes(router fiber.Router, handler *admin.RefundHandler, doc Summary: "恢复退款原审批提交", Description: "仅恢复最新退款审批实例的原提交事件，不创建新审批实例，不直接调用企业微信接口；审批提交结果未知时必须先由恢复任务确认。", Tags: []string{"退款管理"}, Input: new(dto.RefundIDRequest), Output: new(dto.RefundResponse), Auth: true, }) + Register(refund, doc, groupPath, "POST", "/:id/asset-post-processing/retry", handler.RetryAssetPostProcessing, RouteSpec{ + Summary: "人工重试退款资产后处理", Description: "仅平台账号；仅对资产后处理状态为需人工处理（自动重试已达 10 次）的退款单发起，同步执行一次后处理并写尝试记录，返回最新退款详情；仍失败时返回错误并保留需人工处理状态。", Tags: []string{"退款管理"}, Input: new(dto.RefundIDRequest), Output: new(dto.RefundResponse), Auth: true, + }) + Register(refund, doc, groupPath, "POST", "/:id/approve", handler.Approve, RouteSpec{ Summary: "审批通过退款申请", Tags: []string{"退款管理"}, diff --git a/internal/service/iot_card/stop_resume_command.go b/internal/service/iot_card/stop_resume_command.go index 90794be5..ad86f28f 100644 --- a/internal/service/iot_card/stop_resume_command.go +++ b/internal/service/iot_card/stop_resume_command.go @@ -3,6 +3,7 @@ package iot_card import ( "context" stderrors "errors" + "github.com/break/junhong_cmp_fiber/internal/infrastructure/gatewaylimit" "time" "github.com/google/uuid" @@ -52,6 +53,17 @@ func (s *StopResumeService) DecideCommand(ctx context.Context, cardID uint, scen return CommandDecision{Card: card, Action: decision.Action, Reason: decision.Reason}, nil } +// RecordCommandSkipped 为执行前结束、未调用上游的停复机指令写一条"跳过"停复机记录；记录写入失败只告警。 +func (s *StopResumeService) RecordCommandSkipped(ctx context.Context, card *model.IotCard, action, scene, reason, skipReason string) { + if s.stopResumeRecorder == nil || card == nil { + return + } + s.stopResumeRecorder.Record(ctx, gatewaylimit.StopResumeRecordInput{ + CardID: card.ID, ICCID: card.ICCID, CarrierID: card.CarrierID, Action: action, + Source: commandSource(ctx), Scene: scene, Reason: reason, Attempt: 0, SkipReason: skipReason, + }) +} + // ErrCommandBusy 表示该卡复机正在其他路径执行（持单卡复机锁），本次未调用上游。 var ErrCommandBusy = stderrors.New("该卡复机正在执行中") diff --git a/internal/service/refund/approval_decision.go b/internal/service/refund/approval_decision.go index 5fe9fdef..2cbf5ff2 100644 --- a/internal/service/refund/approval_decision.go +++ b/internal/service/refund/approval_decision.go @@ -290,16 +290,4 @@ func (s *Service) ProcessCommissionDeduction(ctx context.Context, refundID uint) return s.clawbackCommission(ctx, refundID) } -func (s *Service) ProcessAssetPostProcessing(ctx context.Context, refundID uint) error { - s.handleRefundAssetProcessing(ctx, refundID) - var refund model.RefundRequest - if err := s.db.WithContext(ctx).Select("asset_reset").First(&refund, refundID).Error; err != nil { - return errors.Wrap(errors.CodeDatabaseError, err, "复核退款资产后处理状态失败") - } - if !refund.AssetReset { - return errors.New(errors.CodeServiceUnavailable, "退款资产后处理尚未完成") - } - return nil -} - var _ approvalapp.BusinessDecisionHandler = (*Service)(nil) diff --git a/internal/service/refund/asset_post_processing.go b/internal/service/refund/asset_post_processing.go new file mode 100644 index 00000000..86f34034 --- /dev/null +++ b/internal/service/refund/asset_post_processing.go @@ -0,0 +1,236 @@ +package refund + +import ( + "context" + stderrors "errors" + "strconv" + "time" + + "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/model/dto" + "github.com/break/junhong_cmp_fiber/pkg/auditcontext" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/errors" + "github.com/break/junhong_cmp_fiber/pkg/middleware" +) + +// assetPostErrorMessageMax 是尝试记录与退款单上错误摘要的最大长度（字符）。 +const assetPostErrorMessageMax = 500 + +// ProcessAssetPostProcessing 是 Outbox 投递（含补偿扫描重投）的退款资产后处理入口。 +// 后处理失败只更新重试状态并返回 nil，由补偿扫描按下次重试时间拉起，不靠 Outbox 自身重投； +// 只有读写退款单本身失败时返回错误，交给 Outbox 重试。 +func (s *Service) ProcessAssetPostProcessing(ctx context.Context, refundID uint) error { + return s.runAssetPostProcessing(ctx, refundID, constants.RefundPostTriggerOutbox) +} + +// RetryAssetPostProcessingByAdmin 是后台人工重试接口：仅平台账号，按详情可见范围定位后重试并返回最新详情。 +func (s *Service) RetryAssetPostProcessingByAdmin(ctx context.Context, refundID uint) (*dto.RefundResponse, error) { + userType := middleware.GetUserTypeFromContext(ctx) + if userType != constants.UserTypeSuperAdmin && userType != constants.UserTypePlatform { + return nil, errors.New(errors.CodeForbidden, "只有平台账号可以重试退款资产后处理") + } + if _, err := s.refundStore.GetByID(ctx, refundID); err != nil { + return nil, errors.New(errors.CodeNotFound, "退款申请不存在") + } + if err := s.RetryAssetPostProcessing(ctx, refundID); err != nil { + return nil, err + } + return s.GetByID(ctx, refundID) +} + +// RetryAssetPostProcessing 是人工重试入口，只允许对"需人工处理"的退款单发起。 +func (s *Service) RetryAssetPostProcessing(ctx context.Context, refundID uint) error { + var refund model.RefundRequest + if err := s.db.WithContext(ctx).Select("id", "status", "asset_reset", "asset_post_status").First(&refund, refundID).Error; err != nil { + if err == gorm.ErrRecordNotFound { + return errors.New(errors.CodeNotFound, "退款单不存在") + } + return errors.Wrap(errors.CodeDatabaseError, err, "查询退款单失败") + } + if refund.AssetReset || refund.AssetPostStatus != model.RefundAssetPostManualRequired { + return errors.New(errors.CodeInvalidStatus, "只有需人工处理的退款单可以重试资产后处理") + } + if err := s.runAssetPostProcessing(ctx, refundID, constants.RefundPostTriggerManual); err != nil { + return err + } + if err := s.db.WithContext(ctx).Select("asset_reset", "asset_post_last_error").First(&refund, refundID).Error; err != nil { + return errors.Wrap(errors.CodeDatabaseError, err, "复核退款资产后处理状态失败") + } + if !refund.AssetReset { + return errors.New(errors.CodeServiceUnavailable, "资产后处理仍未完成："+refund.AssetPostLastError) + } + return nil +} + +func (s *Service) runAssetPostProcessing(ctx context.Context, refundID uint, trigger string) error { + var refund model.RefundRequest + if err := s.db.WithContext(ctx).Where("id = ?", refundID).First(&refund).Error; err != nil { + return errors.Wrap(errors.CodeDatabaseError, err, "查询退款单失败") + } + if refund.AssetReset { + return nil + } + now := time.Now() + if trigger == constants.RefundPostTriggerOutbox { + if refund.AssetPostStatus == model.RefundAssetPostManualRequired { + return nil + } + if refund.AssetPostNextAt != nil && refund.AssetPostNextAt.After(now) { + return nil + } + ctx = auditcontext.With(ctx, auditcontext.Context{ + ActorKind: constants.AuditActorSystemTask, ActorID: constants.AuditActorIDRefundAssetPostProcessing, + ActorName: "退款资产自动后处理任务", Source: constants.AuditSourceWorker, CorrelationID: refund.RefundNo, + }) + } + + // 先占用本次尝试序号，使完成审计能读到总尝试次数。 + attempt := refund.AssetPostAttempts + 1 + if err := s.db.WithContext(ctx).Model(&model.RefundRequest{}).Where("id = ?", refundID). + Update("asset_post_attempts", attempt).Error; err != nil { + return errors.Wrap(errors.CodeDatabaseError, err, "更新退款资产后处理次数失败") + } + refund.AssetPostAttempts = attempt + + outcome := s.handleRefundAssetProcessing(ctx, &refund) + record := s.recordAssetPostAttempt(ctx, &refund, outcome, attempt, trigger, time.Since(now)) + if outcome.Err == nil { + return s.completeAssetPostProcessing(ctx, refundID) + } + return s.failAssetPostProcessing(ctx, &refund, outcome, attempt, trigger, record) +} + +func (s *Service) completeAssetPostProcessing(ctx context.Context, refundID uint) error { + if err := s.db.WithContext(ctx).Model(&model.RefundRequest{}).Where("id = ?", refundID).Updates(map[string]any{ + "asset_post_status": model.RefundAssetPostCompleted, "asset_post_last_error": "", "asset_post_next_at": nil, + }).Error; err != nil { + return errors.Wrap(errors.CodeDatabaseError, err, "更新退款资产后处理状态失败") + } + return nil +} + +func (s *Service) failAssetPostProcessing( + ctx context.Context, refund *model.RefundRequest, outcome refundAssetOutcome, attempt int, trigger string, record *model.RefundPostProcessAttempt, +) error { + message := truncateRunes(outcome.Err.Error(), assetPostErrorMessageMax) + updates := map[string]any{"asset_post_last_error": message} + final := attempt >= constants.RefundPostMaxAttempts + if final { + updates["asset_post_status"] = model.RefundAssetPostManualRequired + updates["asset_post_next_at"] = nil + } else { + updates["asset_post_status"] = model.RefundAssetPostRetryPending + updates["asset_post_next_at"] = time.Now().Add(assetPostBackoff(attempt)) + } + if err := s.db.WithContext(ctx).Model(&model.RefundRequest{}).Where("id = ?", refund.ID).Updates(updates).Error; err != nil { + return errors.Wrap(errors.CodeDatabaseError, err, "更新退款资产后处理重试状态失败") + } + s.recordAssetPostFailureAudit(ctx, refund, outcome, attempt, record, final || trigger == constants.RefundPostTriggerManual) + return nil +} + +// recordAssetPostFailureAudit 写后处理失败：中间失败只写应用日志，转人工或人工重试失败写审计结论。 +func (s *Service) recordAssetPostFailureAudit( + ctx context.Context, refund *model.RefundRequest, outcome refundAssetOutcome, attempt int, record *model.RefundPostProcessAttempt, final bool, +) { + if s.auditWriter == nil || s.db == nil { + return + } + primary := audit.RefundResource(refund, constants.AuditResourceRelationPrimary, constants.AuditResourceRoleRefundTarget) + primary.SubjectVisibility = constants.AuditSubjectInternalOnly + input := audit.AppendInput{ + EventID: "refund:" + strconv.FormatUint(uint64(refund.ID), 10) + ":asset-processed:failed:" + strconv.Itoa(attempt), + ActionCode: constants.AuditActionRefundAssetProcessed, Summary: "退款资产后处理失败（" + outcome.Step + "）", + ScopeType: constants.AuditScopePlatform, CorrelationID: refund.RefundNo, Resources: []audit.ResourceInput{primary}, + Final: final, Attempt: attempt, + Metadata: map[string]any{"asset_post_attempts": attempt, "step": outcome.Step, "manual_required": attempt >= constants.RefundPostMaxAttempts}, + } + if final { + input.Summary = "退款资产后处理重试用尽，转人工处理（" + outcome.Step + "）" + for key, value := range assetPostOutcomeMetadata(ctx, s.db, refund) { + input.Metadata[key] = value + } + } + if record != nil && record.ID > 0 { + input.AttemptRecordID = strconv.FormatUint(uint64(record.ID), 10) + } + s.auditWriter.RecordFailure(ctx, s.db, input, outcome.Err) +} + +// recordAssetPostAttempt 写一条尝试记录；写入失败只告警，不影响后处理结果。 +func (s *Service) recordAssetPostAttempt( + ctx context.Context, refund *model.RefundRequest, outcome refundAssetOutcome, attempt int, trigger string, duration time.Duration, +) *model.RefundPostProcessAttempt { + record := &model.RefundPostProcessAttempt{ + RefundID: refund.ID, RefundNo: refund.RefundNo, Step: outcome.Step, AssetType: outcome.AssetType, AssetID: outcome.AssetID, + Attempt: attempt, Trigger: trigger, Result: constants.RefundPostResultSuccess, DurationMS: int(duration.Milliseconds()), + } + if outcome.AssetType == constants.AssetTypeIotCard && outcome.AssetID > 0 { + var card model.IotCard + if err := s.db.WithContext(ctx).Select("iccid").First(&card, outcome.AssetID).Error; err == nil { + record.ICCID = card.ICCID + } + } + if outcome.Err != nil { + record.Result = constants.RefundPostResultFailed + record.ErrorCode, record.ErrorMessage = assetPostErrorDetail(outcome.Err) + } + if err := s.db.WithContext(ctx).Create(record).Error; err != nil { + s.logger.Warn("写退款资产后处理尝试记录失败", zap.Uint("refund_id", refund.ID), zap.Int("attempt", attempt), zap.Error(err)) + return nil + } + return record +} + +// assetPostOutcomeMetadata 返回结论审计需要的总尝试次数、首次失败时间与最后错误。 +func assetPostOutcomeMetadata(ctx context.Context, db *gorm.DB, refund *model.RefundRequest) map[string]any { + metadata := map[string]any{"asset_post_attempts": refund.AssetPostAttempts} + var first model.RefundPostProcessAttempt + if err := db.WithContext(ctx).Where("refund_id = ? AND result = ?", refund.ID, constants.RefundPostResultFailed). + Order("created_at ASC, id ASC").Limit(1).Find(&first).Error; err == nil && first.ID > 0 { + metadata["first_failed_at"] = first.CreatedAt + } + var last model.RefundPostProcessAttempt + if err := db.WithContext(ctx).Where("refund_id = ? AND result = ?", refund.ID, constants.RefundPostResultFailed). + Order("created_at DESC, id DESC").Limit(1).Find(&last).Error; err == nil && last.ID > 0 { + metadata["last_error_code"] = last.ErrorCode + } + return metadata +} + +// assetPostBackoff 返回第 attempt 次失败后的重试间隔：2^(attempt-1) 分钟，最长 1 小时。 +func assetPostBackoff(attempt int) time.Duration { + if attempt < 1 { + attempt = 1 + } + if attempt > 7 { + return constants.RefundPostMaxBackoff + } + delay := time.Duration(1<<(attempt-1)) * time.Minute + if delay > constants.RefundPostMaxBackoff { + return constants.RefundPostMaxBackoff + } + return delay +} + +func assetPostErrorDetail(err error) (string, string) { + code := strconv.Itoa(errors.CodeInternalError) + var appErr *errors.AppError + if stderrors.As(err, &appErr) && appErr != nil { + code = strconv.Itoa(appErr.Code) + } + return code, truncateRunes(err.Error(), assetPostErrorMessageMax) +} + +func truncateRunes(value string, limit int) string { + runes := []rune(value) + if len(runes) <= limit { + return value + } + return string(runes[:limit]) +} diff --git a/internal/service/refund/audit.go b/internal/service/refund/audit.go index 0c947d38..ba10131a 100644 --- a/internal/service/refund/audit.go +++ b/internal/service/refund/audit.go @@ -71,15 +71,20 @@ func (s *Service) appendRefundAudit( resources = append(resources, chain...) } + metadata := map[string]any{ + "requested_refund_amount": refund.RequestedRefundAmount, + "approved_refund_amount": refund.ApprovedRefundAmount, + } + if actionCode == constants.AuditActionRefundAssetProcessed { + for key, value := range assetPostOutcomeMetadata(ctx, tx, &refund) { + metadata[key] = value + } + } + // 退款审计都是业务结论；对"只写结论"策略的动作（退款资产后处理）标记为最终结论。 return s.auditWriter.Append(ctx, tx, audit.AppendInput{ EventID: eventID, ActionCode: actionCode, Summary: summary, ScopeType: constants.AuditScopePlatform, Result: constants.AuditResultSuccess, - CorrelationID: refund.RefundNo, - Metadata: map[string]any{ - "requested_refund_amount": refund.RequestedRefundAmount, - "approved_refund_amount": refund.ApprovedRefundAmount, - }, - Resources: resources, + CorrelationID: refund.RefundNo, Metadata: metadata, Resources: resources, Final: true, }) } diff --git a/internal/service/refund/service.go b/internal/service/refund/service.go index 6998a2ce..f5320a2f 100644 --- a/internal/service/refund/service.go +++ b/internal/service/refund/service.go @@ -1125,34 +1125,26 @@ func (s *Service) appendWithdrawalRejectAudit( return nil } +// refundAssetOutcome 是一次退款资产后处理的结果：失败时 Step 为失败步骤，Err 非空。 +type refundAssetOutcome struct { + Step string + AssetType string + AssetID uint + Err error +} + // handleRefundAssetProcessing 幂等处理退款后的资产状态。 // 包括退款套餐精准失效、尝试接续待生效主套餐和必要时停机；全部完成后才设置完成标记。 -func (s *Service) handleRefundAssetProcessing(ctx context.Context, refundID uint) { +// 失败不在这里写审计，由 ProcessAssetPostProcessing 写尝试记录、更新重试状态并按"只写结论"写审计。 +func (s *Service) handleRefundAssetProcessing(ctx context.Context, refund *model.RefundRequest) refundAssetOutcome { logger := s.logger - - // 查询退款单 - var refund model.RefundRequest - if err := s.db.Where("id = ?", refundID).First(&refund).Error; err != nil { - logger.Error("退款资产处理：查询退款单失败", zap.Uint("refund_id", refundID), zap.Error(err)) - return - } - if refund.AssetReset { - return - } - ctx = auditcontext.With(ctx, auditcontext.Context{ - ActorKind: constants.AuditActorSystemTask, ActorID: constants.AuditActorIDRefundAssetPostProcessing, - ActorName: "退款资产自动后处理任务", Source: constants.AuditSourceWorker, CorrelationID: refund.RefundNo, - }) + refundID := refund.ID // 查询关联订单 var order model.Order - if err := s.db.Where("id = ?", refund.OrderID).First(&order).Error; err != nil { - s.recordRefundFailure(ctx, constants.AuditActionRefundAssetProcessed, "完成退款资产后处理失败", &refund, nil, err) + if err := s.db.WithContext(ctx).Where("id = ?", refund.OrderID).First(&order).Error; err != nil { logger.Error("退款资产处理：查询订单失败", zap.Uint("refund_id", refundID), zap.Uint("order_id", refund.OrderID), zap.Error(err)) - return - } - recordFailure := func(err error) { - s.recordRefundFailure(ctx, constants.AuditActionRefundAssetProcessed, "完成退款资产后处理失败", &refund, &order, err) + return refundAssetOutcome{Step: constants.RefundPostStepLoad, Err: errors.Wrap(errors.CodeDatabaseError, err, "查询退款关联订单失败")} } // 确定资产类型和 ID @@ -1161,37 +1153,32 @@ func (s *Service) handleRefundAssetProcessing(ctx context.Context, refundID uint switch order.OrderType { case model.OrderTypeSingleCard: if order.IotCardID == nil { - recordFailure(errors.New(errors.CodeInternalError, "退款单卡订单缺少资产ID")) logger.Error("退款资产处理：单卡订单缺少 iot_card_id", zap.Uint("order_id", order.ID)) - return + return refundAssetOutcome{Step: constants.RefundPostStepLoad, Err: errors.New(errors.CodeInternalError, "退款单卡订单缺少资产ID")} } assetType = "iot_card" assetID = *order.IotCardID case model.OrderTypeDevice: if order.DeviceID == nil { - recordFailure(errors.New(errors.CodeInternalError, "退款设备订单缺少资产ID")) logger.Error("退款资产处理：设备订单缺少 device_id", zap.Uint("order_id", order.ID)) - return + return refundAssetOutcome{Step: constants.RefundPostStepLoad, Err: errors.New(errors.CodeInternalError, "退款设备订单缺少资产ID")} } assetType = "device" assetID = *order.DeviceID default: - recordFailure(errors.New(errors.CodeInvalidParam, "退款订单资产类型无效")) logger.Error("退款资产处理：未知订单类型", zap.String("order_type", order.OrderType)) - return + return refundAssetOutcome{Step: constants.RefundPostStepLoad, Err: errors.New(errors.CodeInvalidParam, "退款订单资产类型无效")} + } + fail := func(step string, err error) refundAssetOutcome { + return refundAssetOutcome{Step: step, AssetType: assetType, AssetID: assetID, Err: err} } // 1. 按退款单精准失效套餐（仅处理本次退款订单关联套餐） if s.packageActivationService == nil { - businessErr := errors.New(errors.CodeServiceUnavailable, "退款套餐处理能力未配置") - recordFailure(businessErr) - logger.Error("退款资产处理：套餐激活服务未注入", - zap.Uint("refund_id", refund.ID), - zap.Uint("order_id", order.ID)) - return + logger.Error("退款资产处理：套餐激活服务未注入", zap.Uint("refund_id", refund.ID), zap.Uint("order_id", order.ID)) + return fail(constants.RefundPostStepInvalidate, errors.New(errors.CodeServiceUnavailable, "退款套餐处理能力未配置")) } if err := s.packageActivationService.InvalidatePackagesForRefund(ctx, assetType, assetID, order.ID, refund.ID, refund.RefundNo, refund.PackageUsageID); err != nil { - recordFailure(err) fields := []zap.Field{ zap.String("asset_type", assetType), zap.Uint("asset_id", assetID), @@ -1204,43 +1191,35 @@ func (s *Service) handleRefundAssetProcessing(ctx context.Context, refundID uint fields = append(fields, zap.Uint("package_usage_id", *refund.PackageUsageID)) } logger.Error("退款资产处理：退款套餐精准失效失败", fields...) - return + return fail(constants.RefundPostStepInvalidate, err) } // 2. 尝试按购买顺序接续待生效主套餐 if _, err := s.packageActivationService.ActivateNextPendingMainPackage(ctx, assetType, assetID); err != nil { - recordFailure(err) logger.Error("退款资产处理：接续激活待生效套餐失败", - zap.String("asset_type", assetType), - zap.Uint("asset_id", assetID), - zap.Uint("refund_id", refund.ID), - zap.Error(err)) - return + zap.String("asset_type", assetType), zap.Uint("asset_id", assetID), zap.Uint("refund_id", refund.ID), zap.Error(err)) + return fail(constants.RefundPostStepActivate, err) } hasActiveMain, err := s.packageActivationService.HasActiveMainPackage(ctx, assetType, assetID) if err != nil { - recordFailure(err) logger.Error("退款资产处理：查询生效主套餐失败", - zap.String("asset_type", assetType), - zap.Uint("asset_id", assetID), - zap.Uint("refund_id", refund.ID), - zap.Error(err)) - return + zap.String("asset_type", assetType), zap.Uint("asset_id", assetID), zap.Uint("refund_id", refund.ID), zap.Error(err)) + return fail(constants.RefundPostStepCheckMain, err) } if !hasActiveMain { // 3. 无可用主套餐时才停机；退款不再重置世代或重建钱包。 if !s.stopAsset(ctx, assetType, assetID) { - recordFailure(errors.New(errors.CodeServiceUnavailable, "退款资产停机处理失败")) - return + return fail(constants.RefundPostStepStop, errors.New(errors.CodeServiceUnavailable, "退款资产停机处理失败")) } } // 4. 标记退款后资产处理已完成 if err := s.markRefundAssetProcessed(ctx, refundID); err != nil { - recordFailure(err) logger.Error("退款资产处理：更新处理标记失败", zap.Uint("refund_id", refundID), zap.Error(err)) + return fail(constants.RefundPostStepMark, err) } + return refundAssetOutcome{Step: constants.RefundPostStepDone, AssetType: assetType, AssetID: assetID} } // stopAsset 根据资产类型执行停机操作 diff --git a/internal/task/carrier_stop_resume_executor.go b/internal/task/carrier_stop_resume_executor.go index 14a5b09a..c766ab2d 100644 --- a/internal/task/carrier_stop_resume_executor.go +++ b/internal/task/carrier_stop_resume_executor.go @@ -5,6 +5,7 @@ import ( "encoding/json" stderrors "errors" "strconv" + "strings" "time" "go.uber.org/zap" @@ -25,6 +26,7 @@ type StopResumeCommandService interface { ConfirmCommandState(ctx context.Context, card *model.IotCard, action, reason string) error EvaluateDeviceCommands(ctx context.Context, deviceID uint, scenes []string) error LegacyDeviceOpLocked(ctx context.Context, deviceID uint) bool + RecordCommandSkipped(ctx context.Context, card *model.IotCard, action, scene, reason, skipReason string) } // stopResumeTaskTimeout 是一条停复机指令的总执行时长上限（一次判断 + 一次网关调用）。 @@ -148,6 +150,7 @@ func (e *StopResumeExecutor) execute(ctx context.Context, msg carrierqueue.StopR e.logger.Info("停复机指令执行前重新判断不再需要，结束指令", zap.Uint("card_id", msg.CardID), zap.String("action", cmd.Action), zap.String("decided", decision.Action), zap.Strings("scenes", cmd.Scenes)) e.recovered(cmd, card) + e.service.RecordCommandSkipped(ctx, card, cmd.Action, skipScene(cmd.Scenes), cmd.Reason, skipReason(card, cmd.Action, decision.Action)) return srEnded() } if clearedPending { @@ -330,6 +333,33 @@ func (e *StopResumeExecutor) recovered(cmd carrierqueue.CommandState, card *mode } } +// 停复机指令执行前结束的原因（写入"跳过"停复机记录）。 +const ( + srSkipAlreadyTarget = "执行前已是目标状态" + srSkipConditionNotMet = "执行前判断停复机条件不再满足" + srSkipOppositeAction = "执行前判断需要相反动作" +) + +// skipReason 返回指令执行前结束的原因。 +func skipReason(card *model.IotCard, action, decided string) string { + if card != nil && card.NetworkStatus == targetNetworkStatus(action) { + return srSkipAlreadyTarget + } + if decided != "" { + return srSkipOppositeAction + } + return srSkipConditionNotMet +} + +// skipScene 把指令的场景合并为记录的场景字段（列宽 50）。 +func skipScene(scenes []string) string { + scene := strings.Join(scenes, ",") + if len(scene) > 50 { + return scene[:50] + } + return scene +} + func targetNetworkStatus(action string) int { if action == constants.StopResumeActionStop { return constants.NetworkStatusOffline diff --git a/migrations/000271_add_refund_post_process_attempt.down.sql b/migrations/000271_add_refund_post_process_attempt.down.sql new file mode 100644 index 00000000..9933d275 --- /dev/null +++ b/migrations/000271_add_refund_post_process_attempt.down.sql @@ -0,0 +1,9 @@ +BEGIN; +DROP TABLE IF EXISTS tb_refund_post_process_attempt; +DROP INDEX IF EXISTS idx_refund_request_asset_post_status; +ALTER TABLE tb_refund_request + DROP COLUMN IF EXISTS asset_post_next_at, + DROP COLUMN IF EXISTS asset_post_last_error, + DROP COLUMN IF EXISTS asset_post_attempts, + DROP COLUMN IF EXISTS asset_post_status; +COMMIT; diff --git a/migrations/000271_add_refund_post_process_attempt.up.sql b/migrations/000271_add_refund_post_process_attempt.up.sql new file mode 100644 index 00000000..72fca3d5 --- /dev/null +++ b/migrations/000271_add_refund_post_process_attempt.up.sql @@ -0,0 +1,48 @@ +-- 退款资产后处理：逐次尝试记录 + 退款单上的当前状态、次数、最后错误与下次重试时间。 +-- 每单自动重试最多 10 次，尝试记录量很小，不分区。 +BEGIN; + +ALTER TABLE tb_refund_request + ADD COLUMN IF NOT EXISTS asset_post_status smallint NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS asset_post_attempts int NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS asset_post_last_error varchar(500) NOT NULL DEFAULT '', + ADD COLUMN IF NOT EXISTS asset_post_next_at timestamptz; + +COMMENT ON COLUMN tb_refund_request.asset_post_status IS '资产后处理状态 0-未开始 1-待重试 2-已完成 3-需人工处理'; +COMMENT ON COLUMN tb_refund_request.asset_post_attempts IS '资产后处理已尝试次数'; +COMMENT ON COLUMN tb_refund_request.asset_post_last_error IS '资产后处理最后一次错误摘要'; +COMMENT ON COLUMN tb_refund_request.asset_post_next_at IS '资产后处理下次自动重试时间，为空表示可立即执行'; + +-- 已完成的存量退款单同步为已完成。 +UPDATE tb_refund_request SET asset_post_status = 2 WHERE asset_reset = true AND asset_post_status = 0; + +CREATE INDEX IF NOT EXISTS idx_refund_request_asset_post_status ON tb_refund_request (asset_post_status, asset_post_next_at); + +CREATE TABLE IF NOT EXISTS tb_refund_post_process_attempt ( + id BIGSERIAL PRIMARY KEY, + refund_id BIGINT NOT NULL, + refund_no VARCHAR(64) NOT NULL DEFAULT '', + step VARCHAR(32) NOT NULL, + asset_type VARCHAR(16) NOT NULL DEFAULT '', + asset_id BIGINT NOT NULL DEFAULT 0, + iccid VARCHAR(30) NOT NULL DEFAULT '', + attempt INT NOT NULL, + trigger VARCHAR(16) NOT NULL, + result VARCHAR(16) NOT NULL, + error_code VARCHAR(32) NOT NULL DEFAULT '', + error_message VARCHAR(500) NOT NULL DEFAULT '', + duration_ms INT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +COMMENT ON TABLE tb_refund_post_process_attempt IS '退款资产后处理逐次尝试记录，只新增不修改'; +COMMENT ON COLUMN tb_refund_post_process_attempt.step IS '处理步骤 load-读取订单与资产 invalidate-套餐失效 activate-接续激活 check_main-查询生效主套餐 stop-停机评估或停机 mark-标记完成 done-全部完成'; +COMMENT ON COLUMN tb_refund_post_process_attempt.trigger IS '触发方式 outbox-Outbox投递 manual-人工重试'; +COMMENT ON COLUMN tb_refund_post_process_attempt.result IS '结果 success-成功 failed-失败'; + +CREATE INDEX IF NOT EXISTS idx_refund_post_attempt_refund ON tb_refund_post_process_attempt (refund_id, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_refund_post_attempt_time ON tb_refund_post_process_attempt (created_at DESC); +CREATE INDEX IF NOT EXISTS idx_refund_post_attempt_result ON tb_refund_post_process_attempt (result, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_refund_post_attempt_error ON tb_refund_post_process_attempt (error_code, created_at DESC); + +COMMIT; diff --git a/openspec/changes/refactor-audit-log-persistence/tasks.md b/openspec/changes/refactor-audit-log-persistence/tasks.md index cceeb467..fe857720 100644 --- a/openspec/changes/refactor-audit-log-persistence/tasks.md +++ b/openspec/changes/refactor-audit-log-persistence/tasks.md @@ -13,19 +13,19 @@ ## 2. 退款后处理与停复机跳过记录（第 2 次提交） -- [ ] 2.1 迁移： +- [x] 2.1 迁移： - 新建 `tb_refund_post_process_attempt`，按月分区，并建立索引； - 退款单新增 `asset_post_status`、`asset_post_attempts`、`asset_post_last_error`、`asset_post_next_at`； - 补齐 down 脚本。 -- [ ] 2.2 `internal/service/refund`： +- [x] 2.2 `internal/service/refund`： - `processRefundAsset` 每次执行都写尝试记录（步骤、触发方式、结果、错误码与错误信息、耗时、关联停复机记录）； - 失败时按指数退避（最长 1 小时）更新退款单； - 达到 10 次后置为需人工处理，并写一条 `outcome` 失败结论审计； - 成功时写 `outcome` 成功结论审计，`asset_reset` 同步置为完成。 -- [ ] 2.3 Outbox 处理器与 `RecoverRefundPostProcessing` 在执行前检查状态和 `next_at`：未到期、已完成、需人工处理的都跳过；补偿扫描负责拉起已到期的待重试单。 -- [ ] 2.4 新增 `POST /refunds/:id/asset-post-processing/retry` 人工重试接口，只允许在需人工处理状态下调用，记为人工重试；在路由 `RouteSpec` 中登记，并生成接口文档。 -- [ ] 2.5 停复机指令执行前的重新判断处，按三种结束原因写 `skipped` 停复机记录（已是目标状态、条件不再满足、不允许停复机）；限流推迟不写。 -- [ ] 2.6 构建、校验、提交并推送 8-11。 +- [x] 2.3 Outbox 处理器与 `RecoverRefundPostProcessing` 在执行前检查状态和 `next_at`：未到期、已完成、需人工处理的都跳过；补偿扫描负责拉起已到期的待重试单。 +- [x] 2.4 新增 `POST /refunds/:id/asset-post-processing/retry` 人工重试接口，只允许在需人工处理状态下调用，记为人工重试；在路由 `RouteSpec` 中登记，并生成接口文档。 +- [x] 2.5 停复机指令执行前的重新判断处，按三种结束原因写 `skipped` 停复机记录（已是目标状态、条件不再满足、不允许停复机）；限流推迟不写。 +- [x] 2.6 构建、校验、提交并推送 8-11。 ## 3. 流量采集分表与原始数据（第 3 次提交） diff --git a/pkg/constants/refund_post_process.go b/pkg/constants/refund_post_process.go new file mode 100644 index 00000000..693a144f --- /dev/null +++ b/pkg/constants/refund_post_process.go @@ -0,0 +1,33 @@ +package constants + +import "time" + +// 退款资产后处理步骤（tb_refund_post_process_attempt.step）。 +const ( + RefundPostStepLoad = "load" // 读取订单与资产 + RefundPostStepInvalidate = "invalidate" // 套餐失效 + RefundPostStepActivate = "activate" // 接续激活 + RefundPostStepCheckMain = "check_main" // 查询生效主套餐 + RefundPostStepStop = "stop" // 停机评估或停机 + RefundPostStepMark = "mark" // 标记完成 + RefundPostStepDone = "done" // 全部完成 +) + +// 退款资产后处理触发方式。 +const ( + RefundPostTriggerOutbox = "outbox" // Outbox 投递（含补偿扫描重投） + RefundPostTriggerManual = "manual" // 后台人工重试 +) + +// 退款资产后处理尝试结果。 +const ( + RefundPostResultSuccess = "success" + RefundPostResultFailed = "failed" +) + +const ( + // RefundPostMaxAttempts 是退款资产后处理自动重试上限，达到后转人工处理。 + RefundPostMaxAttempts = 10 + // RefundPostMaxBackoff 是自动重试间隔上限。 + RefundPostMaxBackoff = time.Hour +)