feat(资产钱包自动续费): 新增全局配置、每日扫描续购与可靠复机

- 新增单行配置表 tb_asset_auto_renewal_config 与尝试记录表 tb_asset_auto_renewal_attempt（迁移 000229/000230） - 每日按上海自然日扫描，窗口内以同一资产钱包可用余额续购当前主套餐，资金/订单/套餐/审计同一事务闭合 - 唯一键保证每资产每日至多一次尝试，占位中断由后续扫描收敛，当日不重试 - 四类失败原因向客户与店铺各投递每日至多一条站内通知，并注册通知类型与个人客户白名单 - 续费成功后按条件经 Outbox 可靠投递复机，新增恢复扫描只查询回填，不使用即发即弃调用 - 配置读写仅超级管理员与平台账号，保存记录操作者、前后值快照并登记统一审计 - tasks 7.1–7.15 全部验证通过（本机隔离 PostgreSQL/Redis，零外部渠道调用）
2026-09-17 16:39:26 +08:00
parent 70e6b186df
commit d52be16802
54 changed files with 5164 additions and 97 deletions
--- a/internal/store/postgres/asset_auto_renewal_store.go
+++ b/internal/store/postgres/asset_auto_renewal_store.go
@@ -0,0 +1,314 @@
+package postgres
+
+import (
+	"context"
+	stderrors "errors"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+
+	"github.com/break/junhong_cmp_fiber/internal/model"
+	"github.com/break/junhong_cmp_fiber/pkg/constants"
+	"github.com/break/junhong_cmp_fiber/pkg/errors"
+)
+
+// autoRenewalAttemptConstraint 是自动续费尝试的部分唯一索引名，用于把 23505 精确识别为「当日已尝试」。
+// 部分唯一索引不能与 GORM OnConflict 组合（谓词未声明时无法命中，见 KNOWN-ISSUE-001），
+// 因此占位写入使用显式插入 + 23505 识别。
+const autoRenewalAttemptConstraint = "uq_asset_auto_renewal_attempt_key"
+
+// shanghaiLocation 是自动续费使用的东八区口径：触发日期与跨日比较都以该时区的自然日为准，
+// 与候选查询、尝试唯一键的日期来源保持一致。
+var shanghaiLocation = time.FixedZone("Asia/Shanghai", 8*60*60)
+
+// AssetAutoRenewalConfigStore 是单行自动续费配置的数据访问层。
+//
+// 配置只有一行（主键恒为 1）：读取按主键取，保存必须在调用方事务内先取行锁再写回，
+// 使并发保存串行化而不是静默覆盖。
+type AssetAutoRenewalConfigStore struct {
+	db *gorm.DB
+}
+
+// NewAssetAutoRenewalConfigStore 创建自动续费配置 Store。
+func NewAssetAutoRenewalConfigStore(db *gorm.DB) *AssetAutoRenewalConfigStore {
+	return &AssetAutoRenewalConfigStore{db: db}
+}
+
+// Get 读取唯一一行自动续费配置。
+func (s *AssetAutoRenewalConfigStore) Get(ctx context.Context) (*model.AssetAutoRenewalConfig, error) {
+	var config model.AssetAutoRenewalConfig
+	if err := s.db.WithContext(ctx).First(&config, constants.AssetAutoRenewalConfigSingletonID).Error; err != nil {
+		return nil, err
+	}
+	return &config, nil
+}
+
+// LockInTx 在调用方事务内对配置行加锁读取，作为并发保存的串行化点。
+func (s *AssetAutoRenewalConfigStore) LockInTx(ctx context.Context, tx *gorm.DB) (*model.AssetAutoRenewalConfig, error) {
+	var config model.AssetAutoRenewalConfig
+	if err := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).
+		First(&config, constants.AssetAutoRenewalConfigSingletonID).Error; err != nil {
+		return nil, err
+	}
+	return &config, nil
+}
+
+// SaveInTx 在调用方事务内写回配置的开关、范围、集合、天数、版本与操作者。
+func (s *AssetAutoRenewalConfigStore) SaveInTx(ctx context.Context, tx *gorm.DB, config *model.AssetAutoRenewalConfig, operatorID uint) error {
+	result := tx.WithContext(ctx).Model(&model.AssetAutoRenewalConfig{}).
+		Where("id = ?", constants.AssetAutoRenewalConfigSingletonID).
+		Updates(map[string]any{
+			"enabled":            config.Enabled,
+			"scope":              config.Scope,
+			"package_ids":        config.PackageIDs,
+			"days_before_expiry": config.DaysBeforeExpiry,
+			"config_version":     config.ConfigVersion,
+			"updater":            operatorID,
+			"updated_at":         time.Now(),
+		})
+	if result.Error != nil {
+		return result.Error
+	}
+	if result.RowsAffected == 0 {
+		return errors.New(errors.CodeInternalError, "自动续费配置行不存在")
+	}
+	return nil
+}
+
+// AssetAutoRenewalAttemptStore 是自动续费尝试记录的数据访问层。
+//
+// 占位写入与失败/跳过终态各走独立短事务；成功终态与续费事实同事务更新。当日重复尝试由唯一键
+// 冲突表达，终态写入一律以记录仍处于允许该终态的状态为条件（ENG-CONC-001）。
+type AssetAutoRenewalAttemptStore struct {
+	db *gorm.DB
+}
+
+// NewAssetAutoRenewalAttemptStore 创建自动续费尝试 Store。
+func NewAssetAutoRenewalAttemptStore(db *gorm.DB) *AssetAutoRenewalAttemptStore {
+	return &AssetAutoRenewalAttemptStore{db: db}
+}
+
+// Load 按主键读取尝试记录。
+func (s *AssetAutoRenewalAttemptStore) Load(ctx context.Context, id uint) (*model.AssetAutoRenewalAttempt, error) {
+	var attempt model.AssetAutoRenewalAttempt
+	if err := s.db.WithContext(ctx).First(&attempt, id).Error; err != nil {
+		return nil, err
+	}
+	return &attempt, nil
+}
+
+// CountByAsset 统计该资产截至触发日的尝试条数，用于跨日尝试次数。
+//
+// 触发日期是 DATE 列：比较必须走显式日期参数（?::date），不得把 time.Time 直接与东八区零点比较，
+// 否则会因时区换算差一天。
+func (s *AssetAutoRenewalAttemptStore) CountByAsset(ctx context.Context, assetType string, assetID uint, triggerDate time.Time) (int64, error) {
+	var count int64
+	if err := s.db.WithContext(ctx).Model(&model.AssetAutoRenewalAttempt{}).
+		Where("asset_type = ? AND asset_id = ? AND trigger_date <= ?::date", assetType, assetID, shanghaiDateValue(triggerDate)).
+		Count(&count).Error; err != nil {
+		return 0, errors.Wrap(errors.CodeDatabaseError, err, "统计跨日自动续费尝试次数失败")
+	}
+	return count, nil
+}
+
+// CreatePlaceholder 写入当次尝试占位。
+//
+// 唯一键冲突（该资产当日已有尝试）返回 (false, nil)，由调用方跳过该资产；其余错误原样返回。
+// 单条 INSERT 本身就是原子的，因此这里**不使用显式事务包裹**：既有的保存点写法
+// （internal/application/carrierthreshold/lock_store.go 的 createLockInTx）是为了在**已经处于
+// 事务中**插入时隔离冲突；本方法要么独立执行、要么由调用方决定事务边界，没有需要隔离的外层事务。
+// 反过来，若在事务闭包内吞掉 23505 后 return nil，GORM 会去提交一个已被 PostgreSQL 中止的事务
+// （ErrTxCommitRollback），「当日已尝试」这一可观察结果就永远不会成立。
+func (s *AssetAutoRenewalAttemptStore) CreatePlaceholder(ctx context.Context, attempt *model.AssetAutoRenewalAttempt) (bool, error) {
+	result := s.db.WithContext(ctx).Create(attempt)
+	if result.Error != nil {
+		if isAutoRenewalAttemptConflict(result.Error) {
+			return false, nil
+		}
+		return false, errors.Wrap(errors.CodeDatabaseError, result.Error, "写入自动续费尝试占位失败")
+	}
+	return result.RowsAffected == 1, nil
+}
+
+// FinalizeInTx 在调用方事务内以「记录仍非终态」为条件写入终态。
+// 成功终态与续费事实同事务，因此本方法只接受调用方事务，不自行开启事务。
+func (s *AssetAutoRenewalAttemptStore) FinalizeInTx(ctx context.Context, tx *gorm.DB, attemptID uint, updates map[string]any) (bool, error) {
+	return finalizeAttempt(ctx, tx, attemptID, updates)
+}
+
+// Finalize 以独立短事务写入失败/跳过终态。
+//
+// 该短事务与已回滚的续费事务不共用连接或事务（ENG-TX-001 例外），条件更新依据尝试记录仍非终态。
+func (s *AssetAutoRenewalAttemptStore) Finalize(ctx context.Context, attemptID uint, updates map[string]any) (bool, error) {
+	updated := false
+	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		result, err := finalizeAttempt(ctx, tx, attemptID, updates)
+		if err != nil {
+			return err
+		}
+		updated = result
+		return nil
+	})
+	if err != nil {
+		return false, err
+	}
+	return updated, nil
+}
+
+func finalizeAttempt(ctx context.Context, tx *gorm.DB, attemptID uint, updates map[string]any) (bool, error) {
+	if len(updates) == 0 {
+		return false, errors.New(errors.CodeInvalidParam, "自动续费终态更新内容为空")
+	}
+	updates["updated_at"] = time.Now()
+	result := tx.WithContext(ctx).Model(&model.AssetAutoRenewalAttempt{}).
+		Where("id = ? AND status = ?", attemptID, constants.AssetAutoRenewalAttemptStatusProcessing).
+		Updates(updates)
+	if result.Error != nil {
+		return false, errors.Wrap(errors.CodeDatabaseError, result.Error, "写入自动续费终态失败")
+	}
+	return result.RowsAffected == 1, nil
+}
+
+// ConvergeUnfinished 收敛触发日期早于指定自然日且仍未终态的尝试记录。
+//
+// 占位后进程中断会让记录停在处理中；该记录既不能被当日重试，也不能永久悬空，
+// 因此由后续扫描收敛为「失败 + interrupted」：failure_reason=interrupted 不属于四类通知原因，
+// 收敛过程不发送任何通知（一个从未进入资金事务的占位行不该产生「订单失败」通知），
+// 也不回写任何资金、订单与复机字段。触发日期是 DATE 列，比较走显式日期参数（?::date），
+// 与「当日不重试」使用同一日期口径。返回收敛条数。
+func (s *AssetAutoRenewalAttemptStore) ConvergeUnfinished(ctx context.Context, before time.Time) (int64, error) {
+	result := s.db.WithContext(ctx).Model(&model.AssetAutoRenewalAttempt{}).
+		Where("status = ? AND trigger_date < ?::date", constants.AssetAutoRenewalAttemptStatusProcessing, shanghaiDateValue(before)).
+		Updates(map[string]any{
+			"status":         constants.AssetAutoRenewalAttemptStatusFailed,
+			"failure_reason": constants.AssetAutoRenewalFailureInterrupted,
+			"failure_detail": "上次执行在占位后中断，未进入资金事务即结束；本记录仅作中断收敛，当日不重试",
+			"updated_at":     time.Now(),
+		})
+	if result.Error != nil {
+		return 0, errors.Wrap(errors.CodeDatabaseError, result.Error, "收敛自动续费非终态尝试失败")
+	}
+	return result.RowsAffected, nil
+}
+
+// shanghaiDateValue 把时间格式化为东八区自然日字符串，供 DATE 列比较使用。
+func shanghaiDateValue(value time.Time) string {
+	return value.In(shanghaiLocation).Format("2006-01-02")
+}
+
+// ClaimResumeSubmission 以「已投递且尚未提交」为条件认领复机执行权。
+//
+// 认领成功即取得至多一次的外部调用权；重复投递的 Outbox 事件不会产生第二次调用。
+func (s *AssetAutoRenewalAttemptStore) ClaimResumeSubmission(ctx context.Context, attemptID uint, now time.Time) (bool, error) {
+	result := s.db.WithContext(ctx).Model(&model.AssetAutoRenewalAttempt{}).
+		Where("id = ? AND resume_status = ? AND resume_submitted_at IS NULL",
+			attemptID, constants.AssetAutoRenewalResumeStatusRequested).
+		Updates(map[string]any{"resume_submitted_at": now, "updated_at": now})
+	if result.Error != nil {
+		return false, errors.Wrap(errors.CodeDatabaseError, result.Error, "认领自动续费复机执行权失败")
+	}
+	return result.RowsAffected == 1, nil
+}
+
+// MarkResumeOutcomeInTx 在调用方事务内条件回写复机结果，只接受记录仍处于期望状态时写入。
+//
+// 与失败通知、失败审计同事务：三者要么一起提交，要么一起回滚，避免出现「终态已写、通知永久丢失」。
+func (s *AssetAutoRenewalAttemptStore) MarkResumeOutcomeInTx(
+	ctx context.Context,
+	tx *gorm.DB,
+	attemptID uint,
+	expectedStatuses []int,
+	status int,
+	integrationID, failureReason string,
+) (bool, error) {
+	return markResumeOutcome(ctx, tx, attemptID, expectedStatuses, status, integrationID, failureReason)
+}
+
+// MarkResumeOutcome 条件回写复机结果（单条 UPDATE 自带原子性，无需显式事务）。
+//
+// 供只需要回填、无需同时投递通知的恢复扫描使用。
+func (s *AssetAutoRenewalAttemptStore) MarkResumeOutcome(
+	ctx context.Context,
+	attemptID uint,
+	expectedStatuses []int,
+	status int,
+	integrationID, failureReason string,
+) (bool, error) {
+	return markResumeOutcome(ctx, s.db, attemptID, expectedStatuses, status, integrationID, failureReason)
+}
+
+// markResumeOutcome 是复机结果条件回写的唯一实现。
+// integrationID 与 failureReason 只在非空时覆盖，避免未知结果把已记录的外部交互标识清空。
+func markResumeOutcome(
+	ctx context.Context,
+	handle *gorm.DB,
+	attemptID uint,
+	expectedStatuses []int,
+	status int,
+	integrationID, failureReason string,
+) (bool, error) {
+	updates := map[string]any{"resume_status": status, "updated_at": time.Now()}
+	if integrationID != "" {
+		updates["resume_integration_id"] = integrationID
+	}
+	if failureReason != "" {
+		updates["resume_failure_reason"] = failureReason
+	}
+	if status == constants.AssetAutoRenewalResumeStatusSucceeded {
+		updates["resume_failure_reason"] = ""
+	}
+	result := handle.WithContext(ctx).Model(&model.AssetAutoRenewalAttempt{}).
+		Where("id = ? AND resume_status IN ?", attemptID, expectedStatuses).
+		Updates(updates)
+	if result.Error != nil {
+		return false, errors.Wrap(errors.CodeDatabaseError, result.Error, "回写自动续费复机结果失败")
+	}
+	return result.RowsAffected == 1, nil
+}
+
+// MarkResumeAnomaly 把超过查询窗口仍无法确认的复机结果标记为需人工核对，退出自动扫描。
+func (s *AssetAutoRenewalAttemptStore) MarkResumeAnomaly(ctx context.Context, attemptID uint, reason string) (bool, error) {
+	result := s.db.WithContext(ctx).Model(&model.AssetAutoRenewalAttempt{}).
+		Where("id = ? AND resume_anomaly_flag = ?", attemptID, 0).
+		Updates(map[string]any{
+			"resume_anomaly_flag":   1,
+			"resume_failure_reason": reason,
+			"updated_at":            time.Now(),
+		})
+	if result.Error != nil {
+		return false, errors.Wrap(errors.CodeDatabaseError, result.Error, "标记自动续费复机异常失败")
+	}
+	return result.RowsAffected == 1, nil
+}
+
+// ScanUnresolvedResumes 扫描仍需按只读查询收敛的复机子结果。
+//
+// 范围只有两类：结果未知（需查询确认）与已提交但超过查询窗口仍未回写（进程中断）。
+// 已标记异常的记录退出扫描，不长期重复查询同一笔无法收敛的结果。
+func (s *AssetAutoRenewalAttemptStore) ScanUnresolvedResumes(ctx context.Context, now time.Time, limit int) ([]model.AssetAutoRenewalAttempt, error) {
+	if limit <= 0 {
+		limit = constants.AssetAutoRenewalRecoveryBatchSize
+	}
+	staleBefore := now.Add(-constants.AssetAutoRenewalResumeQueryWindow)
+	var attempts []model.AssetAutoRenewalAttempt
+	if err := s.db.WithContext(ctx).
+		Where("resume_anomaly_flag = ?", 0).
+		Where("resume_status = ? OR (resume_status = ? AND resume_submitted_at IS NOT NULL AND resume_submitted_at <= ?)",
+			constants.AssetAutoRenewalResumeStatusUnknown,
+			constants.AssetAutoRenewalResumeStatusRequested, staleBefore).
+		Order("id ASC").Limit(limit).Find(&attempts).Error; err != nil {
+		return nil, errors.Wrap(errors.CodeDatabaseError, err, "扫描未收敛的自动续费复机结果失败")
+	}
+	return attempts, nil
+}
+
+// isAutoRenewalAttemptConflict 判断错误是否为尝试唯一键冲突，即「该资产当日已尝试」。
+func isAutoRenewalAttemptConflict(err error) bool {
+	var pgErr *pgconn.PgError
+	if !stderrors.As(err, &pgErr) {
+		return false
+	}
+	return pgErr.Code == "23505" && pgErr.ConstraintName == autoRenewalAttemptConstraint
+}
--- a/internal/store/postgres/asset_wallet_store.go
+++ b/internal/store/postgres/asset_wallet_store.go
@@ -9,6 +9,7 @@ import (
 	"github.com/break/junhong_cmp_fiber/pkg/middleware"
 	"github.com/redis/go-redis/v9"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )

 // AssetWalletStore 资产钱包数据访问层
@@ -61,6 +62,18 @@ func (s *AssetWalletStore) CreateWithTx(ctx context.Context, tx *gorm.DB, wallet
 	return tx.WithContext(ctx).Create(wallet).Error
 }

+// LockByIDWithTx 在调用方事务内按主键加行锁读取资产钱包。
+//
+// 供需要在扣款前串行化同一资产钱包的写用例复用：锁定行后重读余额与状态，
+// 再走既有 DeductBalanceWithTx 的乐观版本条件更新。本方法只加行锁，不修改任何字段与既有约束。
+func (s *AssetWalletStore) LockByIDWithTx(ctx context.Context, tx *gorm.DB, id uint) (*model.AssetWallet, error) {
+	var wallet model.AssetWallet
+	if err := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).First(&wallet, id).Error; err != nil {
+		return nil, err
+	}
+	return &wallet, nil
+}
+
 // DeductBalanceWithTx 扣款（带事务，使用乐观锁）
 func (s *AssetWalletStore) DeductBalanceWithTx(ctx context.Context, tx *gorm.DB, walletID uint, amount int64, version int) error {
 	// 使用乐观锁，检查可用余额是否充足