feat: 退款资产后处理逐次尝试记录与重试上限,停复机指令调用前结束写跳过记录

- 新增 tb_refund_post_process_attempt,每次尝试一条(步骤、触发方式、结果、错误码与信息、耗时)
- 退款单增加 asset_post_status/attempts/last_error/next_at;失败按 2^(n-1) 分钟退避(最长 1 小时),10 次后转人工
- 中间失败只写应用日志,成功或转人工写一条结论审计(带总尝试次数、首次失败时间、最后错误码)
- 补偿扫描跳过需人工处理与未到重试时间的退款单,Outbox 不再无限重投
- 新增 POST /refunds/:id/asset-post-processing/retry 人工重试
- 停复机指令执行前结束(已是目标状态、条件不再满足、需相反动作)写 skipped 停复机记录
- 迁移 000271 需维护者在测试库执行后再部署

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
邹进
2026-11-05 18:33:32 +09:00
parent b7a7430cc3
commit 8eb7d403c4
15 changed files with 488 additions and 77 deletions

View File

@@ -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 {

View File

@@ -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

View File

@@ -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

View File

@@ -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"
}

View File

@@ -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{"退款管理"},

View File

@@ -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("该卡复机正在执行中")

View File

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

View File

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

View File

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

View File

@@ -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 根据资产类型执行停机操作

View File

@@ -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

View File

@@ -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;

View File

@@ -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;

View File

@@ -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 次提交)

View File

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