huang
8d69858413
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
403 lines
13 KiB
Go
403 lines
13 KiB
Go
package packagepkg
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
// ResumeCallback 复机回调接口
|
||
// 用于在套餐激活或流量重置后触发自动复机
|
||
type ResumeCallback interface {
|
||
ResumeCardIfStopped(ctx context.Context, carrierType string, carrierID uint) error
|
||
}
|
||
|
||
type ActivationService struct {
|
||
db *gorm.DB
|
||
redis *redis.Client
|
||
packageUsageStore *postgres.PackageUsageStore
|
||
packageStore *postgres.PackageStore
|
||
packageUsageDailyRecord *postgres.PackageUsageDailyRecordStore
|
||
logger *zap.Logger
|
||
resumeCallback ResumeCallback // 复机回调，可选
|
||
}
|
||
|
||
func NewActivationService(
|
||
db *gorm.DB,
|
||
redis *redis.Client,
|
||
packageUsageStore *postgres.PackageUsageStore,
|
||
packageStore *postgres.PackageStore,
|
||
packageUsageDailyRecord *postgres.PackageUsageDailyRecordStore,
|
||
logger *zap.Logger,
|
||
) *ActivationService {
|
||
return &ActivationService{
|
||
db: db,
|
||
redis: redis,
|
||
packageUsageStore: packageUsageStore,
|
||
packageStore: packageStore,
|
||
packageUsageDailyRecord: packageUsageDailyRecord,
|
||
logger: logger,
|
||
}
|
||
}
|
||
|
||
// SetResumeCallback 任务 24.7: 设置复机回调
|
||
// 在应用启动时由 bootstrap 调用，注入停复机服务
|
||
func (s *ActivationService) SetResumeCallback(callback ResumeCallback) {
|
||
s.resumeCallback = callback
|
||
}
|
||
|
||
// ActivateByRealname 任务 9.2-9.3: 首次实名激活
|
||
// 当用户完成实名后，激活所有待实名激活的套餐
|
||
func (s *ActivationService) ActivateByRealname(ctx context.Context, carrierType string, carrierID uint) error {
|
||
// 查询待实名激活的套餐
|
||
var pendingUsages []*model.PackageUsage
|
||
query := s.db.WithContext(ctx).
|
||
Where("pending_realname_activation = ?", true).
|
||
Where("status = ?", constants.PackageUsageStatusPending)
|
||
|
||
if carrierType == "iot_card" {
|
||
query = query.Where("iot_card_id = ?", carrierID)
|
||
} else if carrierType == "device" {
|
||
query = query.Where("device_id = ?", carrierID)
|
||
} else {
|
||
return errors.New(errors.CodeInvalidParam, "无效的载体类型")
|
||
}
|
||
|
||
if err := query.Order("priority ASC").Find(&pendingUsages).Error; err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "查询待实名激活套餐失败")
|
||
}
|
||
|
||
if len(pendingUsages) == 0 {
|
||
s.logger.Info("没有待实名激活的套餐", zap.String("carrier_type", carrierType), zap.Uint("carrier_id", carrierID))
|
||
return nil
|
||
}
|
||
|
||
now := time.Now()
|
||
|
||
// 在事务中激活套餐
|
||
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
for _, usage := range pendingUsages {
|
||
// 查询套餐信息
|
||
var pkg model.Package
|
||
if err := tx.First(&pkg, usage.PackageID).Error; err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "查询套餐信息失败")
|
||
}
|
||
|
||
// 检查是否是主套餐
|
||
if usage.MasterUsageID == nil {
|
||
// 主套餐：需要检查是否有已激活的主套餐
|
||
var activeMain model.PackageUsage
|
||
err := tx.Where("status = ?", constants.PackageUsageStatusActive).
|
||
Where("master_usage_id IS NULL").
|
||
Where(carrierType+"_id = ?", carrierID).
|
||
Order("priority ASC").
|
||
First(&activeMain).Error
|
||
|
||
if err == nil {
|
||
// 已有激活的主套餐，保持排队状态
|
||
s.logger.Warn("已有激活主套餐，跳过激活",
|
||
zap.Uint("usage_id", usage.ID),
|
||
zap.Uint("active_main_id", activeMain.ID))
|
||
continue
|
||
}
|
||
if err != gorm.ErrRecordNotFound {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "检查生效中主套餐失败")
|
||
}
|
||
}
|
||
|
||
// REALNAME-04: 按 ExpiryBase 选择激活计时基准
|
||
// from_purchase：购买时起算，用 PackageUsage 创建时间；否则用实名触发当前时刻
|
||
var activatedAt time.Time
|
||
if pkg.ExpiryBase == "from_purchase" {
|
||
activatedAt = usage.CreatedAt
|
||
} else {
|
||
activatedAt = now
|
||
}
|
||
expiresAt := CalculateExpiryTime(pkg.CalendarType, activatedAt, pkg.DurationMonths, pkg.DurationDays)
|
||
|
||
// 计算下次重置时间
|
||
nextResetAt := CalculateNextResetTime(pkg.DataResetCycle, pkg.CalendarType, now, activatedAt)
|
||
|
||
// 更新套餐使用记录
|
||
updates := map[string]interface{}{
|
||
"status": constants.PackageUsageStatusActive,
|
||
"pending_realname_activation": false,
|
||
"activated_at": activatedAt,
|
||
"expires_at": expiresAt,
|
||
}
|
||
if nextResetAt != nil {
|
||
updates["next_reset_at"] = *nextResetAt
|
||
}
|
||
|
||
if err := tx.Model(usage).Updates(updates).Error; err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "激活套餐失败")
|
||
}
|
||
|
||
s.syncCarrierStatusActivated(ctx, tx, usage, carrierType, carrierID)
|
||
|
||
s.logger.Info("套餐已激活",
|
||
zap.Uint("usage_id", usage.ID),
|
||
zap.Uint("package_id", usage.PackageID),
|
||
zap.Time("activated_at", activatedAt),
|
||
zap.Time("expires_at", expiresAt))
|
||
|
||
if s.resumeCallback != nil {
|
||
go func(ct string, cid uint) {
|
||
if err := s.resumeCallback.ResumeCardIfStopped(context.Background(), ct, cid); err != nil {
|
||
s.logger.Error("实名激活后自动复机失败",
|
||
zap.String("carrier_type", ct),
|
||
zap.Uint("carrier_id", cid),
|
||
zap.Error(err))
|
||
}
|
||
}(carrierType, carrierID)
|
||
}
|
||
}
|
||
|
||
return nil
|
||
})
|
||
}
|
||
|
||
// ActivateQueuedPackage 任务 9.4-9.7: 排队主套餐激活
|
||
// 当主套餐过期后，激活下一个待生效的主套餐
|
||
func (s *ActivationService) ActivateQueuedPackage(ctx context.Context, carrierType string, carrierID uint) error {
|
||
// 使用 Redis 分布式锁避免并发
|
||
lockKey := constants.RedisPackageActivationLockKey(carrierType, carrierID)
|
||
lockValue := time.Now().String()
|
||
locked, err := s.redis.SetNX(ctx, lockKey, lockValue, 30*time.Second).Result()
|
||
if err != nil {
|
||
return errors.Wrap(errors.CodeRedisError, err, "获取分布式锁失败")
|
||
}
|
||
if !locked {
|
||
s.logger.Warn("套餐激活正在进行中，跳过",
|
||
zap.String("carrier_type", carrierType),
|
||
zap.Uint("carrier_id", carrierID))
|
||
return nil
|
||
}
|
||
defer s.redis.Del(ctx, lockKey)
|
||
|
||
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
// 任务 9.5: 检测并标记过期的主套餐
|
||
now := time.Now()
|
||
var expiredMainUsages []*model.PackageUsage
|
||
err := tx.Where("status = ?", constants.PackageUsageStatusActive).
|
||
Where("master_usage_id IS NULL").
|
||
Where("expires_at <= ?", now).
|
||
Where(carrierType+"_id = ?", carrierID).
|
||
Find(&expiredMainUsages).Error
|
||
|
||
if err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "查询过期主套餐失败")
|
||
}
|
||
|
||
for _, expiredMain := range expiredMainUsages {
|
||
// 更新主套餐状态为已过期
|
||
if err := tx.Model(expiredMain).Update("status", constants.PackageUsageStatusExpired).Error; err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "更新过期主套餐状态失败")
|
||
}
|
||
|
||
s.logger.Info("主套餐已过期",
|
||
zap.Uint("usage_id", expiredMain.ID),
|
||
zap.Time("expires_at", expiredMain.ExpiresAt))
|
||
|
||
// 任务 9.7: 加油包级联失效
|
||
if err := s.invalidateAddons(ctx, tx, expiredMain.ID); err != nil {
|
||
return err
|
||
}
|
||
|
||
// 任务 9.6: 激活下一个待生效主套餐
|
||
if err := s.activateNextMainPackage(ctx, tx, carrierType, carrierID, now); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
return nil
|
||
})
|
||
}
|
||
|
||
// invalidateAddons 任务 9.7: 加油包级联失效
|
||
func (s *ActivationService) invalidateAddons(ctx context.Context, tx *gorm.DB, masterUsageID uint) error {
|
||
var addons []*model.PackageUsage
|
||
if err := tx.Where("master_usage_id = ?", masterUsageID).
|
||
Where("status IN ?", []int{constants.PackageUsageStatusActive, constants.PackageUsageStatusPending}).
|
||
Find(&addons).Error; err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "查询加油包失败")
|
||
}
|
||
|
||
if len(addons) == 0 {
|
||
return nil
|
||
}
|
||
|
||
addonIDs := make([]uint, len(addons))
|
||
for i, addon := range addons {
|
||
addonIDs[i] = addon.ID
|
||
}
|
||
|
||
// 批量更新加油包状态为已失效
|
||
if err := tx.Model(&model.PackageUsage{}).
|
||
Where("id IN ?", addonIDs).
|
||
Update("status", constants.PackageUsageStatusInvalidated).Error; err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "批量失效加油包失败")
|
||
}
|
||
|
||
s.logger.Info("加油包已级联失效",
|
||
zap.Uint("master_usage_id", masterUsageID),
|
||
zap.Int("addon_count", len(addons)))
|
||
|
||
return nil
|
||
}
|
||
|
||
// activateNextMainPackage 任务 9.6: 激活下一个待生效主套餐
|
||
func (s *ActivationService) activateNextMainPackage(ctx context.Context, tx *gorm.DB, carrierType string, carrierID uint, now time.Time) error {
|
||
// 查询下一个待生效主套餐
|
||
var nextMain model.PackageUsage
|
||
err := tx.Where("status = ?", constants.PackageUsageStatusPending).
|
||
Where("master_usage_id IS NULL").
|
||
Where(carrierType+"_id = ?", carrierID).
|
||
Order("priority ASC").
|
||
First(&nextMain).Error
|
||
|
||
if err == gorm.ErrRecordNotFound {
|
||
s.logger.Info("没有待生效的主套餐",
|
||
zap.String("carrier_type", carrierType),
|
||
zap.Uint("carrier_id", carrierID))
|
||
return nil
|
||
}
|
||
if err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "查询下一个待生效主套餐失败")
|
||
}
|
||
|
||
// 查询套餐信息
|
||
var pkg model.Package
|
||
if err := tx.First(&pkg, nextMain.PackageID).Error; err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "查询套餐信息失败")
|
||
}
|
||
|
||
// 激活套餐
|
||
activatedAt := now
|
||
expiresAt := CalculateExpiryTime(pkg.CalendarType, activatedAt, pkg.DurationMonths, pkg.DurationDays)
|
||
|
||
// 计算下次重置时间
|
||
nextResetAt := CalculateNextResetTime(pkg.DataResetCycle, pkg.CalendarType, now, activatedAt)
|
||
|
||
// 更新套餐使用记录
|
||
updates := map[string]interface{}{
|
||
"status": constants.PackageUsageStatusActive,
|
||
"activated_at": activatedAt,
|
||
"expires_at": expiresAt,
|
||
}
|
||
if nextResetAt != nil {
|
||
updates["next_reset_at"] = *nextResetAt
|
||
}
|
||
|
||
if err := tx.Model(&nextMain).Updates(updates).Error; err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "激活排队主套餐失败")
|
||
}
|
||
|
||
s.syncCarrierStatusActivated(ctx, tx, &nextMain, carrierType, carrierID)
|
||
|
||
s.logger.Info("排队主套餐已激活",
|
||
zap.Uint("usage_id", nextMain.ID),
|
||
zap.Uint("package_id", nextMain.PackageID),
|
||
zap.Time("activated_at", activatedAt),
|
||
zap.Time("expires_at", expiresAt))
|
||
|
||
if s.resumeCallback != nil {
|
||
go func(ct string, cid uint) {
|
||
if err := s.resumeCallback.ResumeCardIfStopped(context.Background(), ct, cid); err != nil {
|
||
s.logger.Error("排队激活后自动复机失败",
|
||
zap.String("carrier_type", ct),
|
||
zap.Uint("carrier_id", cid),
|
||
zap.Error(err))
|
||
}
|
||
}(carrierType, carrierID)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (s *ActivationService) syncCarrierStatusActivated(ctx context.Context, tx *gorm.DB, usage *model.PackageUsage, carrierType string, carrierID uint) {
|
||
if usage == nil {
|
||
return
|
||
}
|
||
|
||
if usage.UsageType == constants.PackageUsageTypeDevice || usage.DeviceID > 0 || carrierType == constants.AssetTypeDevice {
|
||
deviceID := usage.DeviceID
|
||
if deviceID == 0 {
|
||
deviceID = carrierID
|
||
}
|
||
if deviceID == 0 {
|
||
return
|
||
}
|
||
|
||
if err := tx.WithContext(ctx).
|
||
Model(&model.Device{}).
|
||
Where("id = ? AND status < ?", deviceID, constants.DeviceStatusActivated).
|
||
Update("status", constants.DeviceStatusActivated).Error; err != nil {
|
||
s.logger.Warn("套餐激活后更新设备状态失败",
|
||
zap.Uint("device_id", deviceID),
|
||
zap.Uint("usage_id", usage.ID),
|
||
zap.Error(err))
|
||
}
|
||
return
|
||
}
|
||
|
||
cardID := usage.IotCardID
|
||
if cardID == 0 {
|
||
cardID = carrierID
|
||
}
|
||
if cardID == 0 {
|
||
return
|
||
}
|
||
|
||
if err := tx.WithContext(ctx).
|
||
Model(&model.IotCard{}).
|
||
Where("id = ? AND status < ?", cardID, constants.IotCardStatusActivated).
|
||
Update("status", constants.IotCardStatusActivated).Error; err != nil {
|
||
s.logger.Warn("套餐激活后更新卡状态失败",
|
||
zap.Uint("iot_card_id", cardID),
|
||
zap.Uint("usage_id", usage.ID),
|
||
zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// InvalidateAllPackagesByAsset 批量失效资产关联的所有有效套餐
|
||
// 退款时调用：将该资产下状态为待生效(0)、生效中(1)、已用完(2)的套餐全部标记为已失效(4)
|
||
func (s *ActivationService) InvalidateAllPackagesByAsset(ctx context.Context, assetType string, assetID uint) error {
|
||
query := s.db.WithContext(ctx).
|
||
Model(&model.PackageUsage{}).
|
||
Where("status IN ?", []int{
|
||
constants.PackageUsageStatusPending,
|
||
constants.PackageUsageStatusActive,
|
||
constants.PackageUsageStatusDepleted,
|
||
})
|
||
|
||
switch assetType {
|
||
case "iot_card":
|
||
query = query.Where("iot_card_id = ?", assetID)
|
||
case "device":
|
||
query = query.Where("device_id = ?", assetID)
|
||
default:
|
||
return errors.New(errors.CodeInvalidParam, "无效的资产类型")
|
||
}
|
||
|
||
result := query.Update("status", constants.PackageUsageStatusInvalidated)
|
||
if result.Error != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, result.Error, "批量失效套餐失败")
|
||
}
|
||
|
||
s.logger.Info("批量失效套餐完成",
|
||
zap.String("asset_type", assetType),
|
||
zap.Uint("asset_id", assetID),
|
||
zap.Int64("affected", result.RowsAffected),
|
||
)
|
||
|
||
return nil
|
||
}
|