shanshanzhong
c582087c0f
Some checks failed
Build docker and publish / build (20.15.1) (push) Failing after 6m27s
feat: 添加版本和构建时间变量 fix: 修正短信队列类型注释错误 style: 清理未使用的代码和测试文件 docs: 更新安装文档中的下载链接 chore: 迁移数据库脚本添加日志和订阅配置
408 lines
12 KiB
Go
408 lines
12 KiB
Go
package task
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strconv"
|
||
"time"
|
||
|
||
"github.com/hibiken/asynq"
|
||
"github.com/perfect-panel/server/internal/model/log"
|
||
"github.com/perfect-panel/server/internal/model/task"
|
||
"github.com/perfect-panel/server/internal/model/user"
|
||
"github.com/perfect-panel/server/internal/svc"
|
||
"github.com/perfect-panel/server/pkg/logger"
|
||
"github.com/perfect-panel/server/pkg/tool"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
const (
|
||
UnitTimeNoLimit = "NoLimit" // Unlimited time subscription
|
||
UnitTimeYear = "Year" // Annual subscription
|
||
UnitTimeMonth = "Month" // Monthly subscription
|
||
UnitTimeDay = "Day" // Daily subscription
|
||
UnitTimeHour = "Hour" // Hourly subscription
|
||
UnitTimeMinute = "Minute" // Per-minute subscription
|
||
|
||
)
|
||
|
||
type QuotaTaskLogic struct {
|
||
svcCtx *svc.ServiceContext
|
||
}
|
||
|
||
type ErrorInfo struct {
|
||
UserSubscribeId int64 `json:"user_subscribe_id"`
|
||
Error string `json:"error"`
|
||
}
|
||
|
||
func NewQuotaTaskLogic(svcCtx *svc.ServiceContext) *QuotaTaskLogic {
|
||
return &QuotaTaskLogic{
|
||
svcCtx: svcCtx,
|
||
}
|
||
}
|
||
|
||
func (l *QuotaTaskLogic) ProcessTask(ctx context.Context, t *asynq.Task) error {
|
||
taskID, err := l.parseTaskID(ctx, t.Payload())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
taskInfo, err := l.getTaskInfo(ctx, taskID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
if taskInfo.Status != 0 {
|
||
logger.WithContext(ctx).Info("[QuotaTaskLogic.ProcessTask] task already processed",
|
||
logger.Field("taskID", taskID),
|
||
logger.Field("status", taskInfo.Status),
|
||
)
|
||
return nil
|
||
}
|
||
|
||
scope, content, err := l.parseTaskData(ctx, taskInfo)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
subscribes, err := l.getSubscribes(ctx, scope.Objects)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err = l.processSubscribes(ctx, subscribes, content, taskInfo); err != nil {
|
||
return err
|
||
}
|
||
// 清理用户缓存（仅在有赠送金时清理）
|
||
if content.GiftValue != 0 {
|
||
var userIds []int64
|
||
for _, sub := range subscribes {
|
||
userIds = append(userIds, sub.UserId)
|
||
}
|
||
userIds = tool.RemoveDuplicateElements(userIds...)
|
||
var users []*user.User
|
||
if err = l.svcCtx.DB.WithContext(ctx).Model(&user.User{}).Where("id IN ?", userIds).Find(&users).Error; err != nil {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.ProcessTask] find users error",
|
||
logger.Field("error", err.Error()),
|
||
logger.Field("userIDs", userIds))
|
||
}
|
||
err = l.svcCtx.UserModel.ClearUserCache(ctx, users...)
|
||
if err != nil {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.ProcessTask] clear user cache error",
|
||
logger.Field("error", err.Error()),
|
||
logger.Field("userIDs", userIds))
|
||
}
|
||
}
|
||
|
||
// 清理用户订阅缓存
|
||
err = l.svcCtx.UserModel.ClearSubscribeCache(ctx, subscribes...)
|
||
if err != nil {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.ProcessTask] clear subscribe cache error",
|
||
logger.Field("error", err.Error()))
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (l *QuotaTaskLogic) parseTaskID(ctx context.Context, payload []byte) (int64, error) {
|
||
if len(payload) == 0 {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.parseTaskID] empty payload")
|
||
return 0, asynq.SkipRetry
|
||
}
|
||
|
||
taskID, err := strconv.ParseInt(string(payload), 10, 64)
|
||
if err != nil {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.parseTaskID] invalid task ID",
|
||
logger.Field("error", err.Error()),
|
||
logger.Field("payload", string(payload)),
|
||
)
|
||
return 0, asynq.SkipRetry
|
||
}
|
||
return taskID, nil
|
||
}
|
||
|
||
func (l *QuotaTaskLogic) getTaskInfo(ctx context.Context, taskID int64) (*task.Task, error) {
|
||
var taskInfo *task.Task
|
||
if err := l.svcCtx.DB.WithContext(ctx).Model(&task.Task{}).Where("id = ?", taskID).First(&taskInfo).Error; err != nil {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.getTaskInfo] find task error",
|
||
logger.Field("error", err.Error()),
|
||
logger.Field("taskID", taskID),
|
||
)
|
||
return nil, asynq.SkipRetry
|
||
}
|
||
return taskInfo, nil
|
||
}
|
||
|
||
func (l *QuotaTaskLogic) parseTaskData(ctx context.Context, taskInfo *task.Task) (task.QuotaScope, task.QuotaContent, error) {
|
||
var scope task.QuotaScope
|
||
if err := scope.Unmarshal([]byte(taskInfo.Scope)); err != nil {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.parseTaskData] unmarshal scope error",
|
||
logger.Field("error", err.Error()),
|
||
)
|
||
return scope, task.QuotaContent{}, asynq.SkipRetry
|
||
}
|
||
|
||
var content task.QuotaContent
|
||
if err := content.Unmarshal([]byte(taskInfo.Content)); err != nil {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.parseTaskData] unmarshal content error",
|
||
logger.Field("error", err.Error()),
|
||
)
|
||
return scope, content, asynq.SkipRetry
|
||
}
|
||
return scope, content, nil
|
||
}
|
||
|
||
func (l *QuotaTaskLogic) getSubscribes(ctx context.Context, subscriberIDs []int64) ([]*user.Subscribe, error) {
|
||
var subscribes []*user.Subscribe
|
||
if err := l.svcCtx.DB.WithContext(ctx).Model(&user.Subscribe{}).Where("id IN ?", subscriberIDs).Find(&subscribes).Error; err != nil {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.getSubscribes] find subscribes error",
|
||
logger.Field("error", err.Error()),
|
||
logger.Field("subscribers", subscriberIDs),
|
||
)
|
||
return nil, asynq.SkipRetry
|
||
}
|
||
return subscribes, nil
|
||
}
|
||
|
||
func (l *QuotaTaskLogic) processSubscribes(ctx context.Context, subscribes []*user.Subscribe, content task.QuotaContent, taskInfo *task.Task) error {
|
||
tx := l.svcCtx.DB.WithContext(ctx).Begin()
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
tx.Rollback()
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.processSubscribes] transaction panic",
|
||
logger.Field("panic", r),
|
||
)
|
||
}
|
||
}()
|
||
|
||
var errors []ErrorInfo
|
||
now := time.Now()
|
||
|
||
for _, sub := range subscribes {
|
||
if err := l.processSubscription(tx, sub, content, now, &errors); err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
}
|
||
|
||
// 根据错误情况决定任务状态
|
||
status := int8(2) // Completed
|
||
if len(errors) > 0 {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.processSubscribes] some subscriptions failed",
|
||
logger.Field("total", len(subscribes)),
|
||
logger.Field("failed", len(errors)),
|
||
)
|
||
// 如果所有订阅都失败，标记为失败状态
|
||
if len(errors) == len(subscribes) {
|
||
status = 3 // Failed
|
||
}
|
||
errs, err := json.Marshal(errors)
|
||
if err != nil {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.processSubscribes] marshal errors failed",
|
||
logger.Field("error", err.Error()),
|
||
)
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
taskInfo.Errors = string(errs)
|
||
}
|
||
|
||
taskInfo.Current = uint64(len(subscribes))
|
||
taskInfo.Status = status
|
||
err := tx.Where("id = ?", taskInfo.Id).Save(taskInfo).Error
|
||
if err != nil {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.processSubscribes] update task status error",
|
||
logger.Field("error", err.Error()),
|
||
logger.Field("taskID", taskInfo.Id),
|
||
)
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
|
||
if err = tx.Commit().Error; err != nil {
|
||
logger.WithContext(ctx).Error("[QuotaTaskLogic.processSubscribes] commit transaction error",
|
||
logger.Field("error", err.Error()),
|
||
)
|
||
return err
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (l *QuotaTaskLogic) processSubscription(tx *gorm.DB, sub *user.Subscribe, content task.QuotaContent, now time.Time, errors *[]ErrorInfo) error {
|
||
// 验证订阅数据
|
||
if sub == nil {
|
||
*errors = append(*errors, ErrorInfo{
|
||
UserSubscribeId: 0,
|
||
Error: "subscription is nil",
|
||
})
|
||
return nil
|
||
}
|
||
|
||
updated := false
|
||
|
||
// 处理时间延长 - 修复逻辑：只要Days不为0就处理，不管ExpireTime是否为0
|
||
if content.Days != 0 {
|
||
if sub.ExpireTime.Unix() == 0 || sub.ExpireTime.Before(now) {
|
||
// 如果没有过期时间或已过期，从现在开始计算
|
||
sub.ExpireTime = now.AddDate(0, 0, int(content.Days))
|
||
} else {
|
||
// 在原有过期时间基础上延长
|
||
sub.ExpireTime = sub.ExpireTime.AddDate(0, 0, int(content.Days))
|
||
}
|
||
// 如果订阅延长到未来时间，设置为激活状态
|
||
if sub.ExpireTime.After(now) && sub.Status != 1 {
|
||
sub.Status = 1 // Active
|
||
}
|
||
updated = true
|
||
}
|
||
|
||
// 处理流量重置
|
||
if content.ResetTraffic {
|
||
sub.Download = 0
|
||
sub.Upload = 0
|
||
updated = true
|
||
if err := l.createResetTrafficLog(tx, sub.Id, sub.UserId, now); err != nil {
|
||
// 记录错误但不阻断整个任务,日志失败不影响主流程
|
||
*errors = append(*errors, ErrorInfo{
|
||
UserSubscribeId: sub.Id,
|
||
Error: "create reset traffic log error: " + err.Error(),
|
||
})
|
||
}
|
||
}
|
||
|
||
// 处理赠送金
|
||
if content.GiftValue != 0 {
|
||
if err := l.processGift(tx, sub, content, now, errors); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
// 只有在有更新时才保存订阅信息
|
||
if updated {
|
||
if err := tx.Where("id = ?", sub.Id).Save(sub).Error; err != nil {
|
||
*errors = append(*errors, ErrorInfo{
|
||
UserSubscribeId: sub.Id,
|
||
Error: "update subscription error: " + err.Error(),
|
||
})
|
||
return nil
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (l *QuotaTaskLogic) processGift(tx *gorm.DB, sub *user.Subscribe, content task.QuotaContent, now time.Time, errors *[]ErrorInfo) error {
|
||
// 验证赠送类型
|
||
if content.GiftType != 1 && content.GiftType != 2 {
|
||
*errors = append(*errors, ErrorInfo{
|
||
UserSubscribeId: sub.Id,
|
||
Error: fmt.Sprintf("invalid gift type: %d", content.GiftType),
|
||
})
|
||
return nil
|
||
}
|
||
|
||
var userInfo user.User
|
||
if err := tx.Model(&user.User{}).Where("id = ?", sub.UserId).First(&userInfo).Error; err != nil {
|
||
*errors = append(*errors, ErrorInfo{
|
||
UserSubscribeId: sub.Id,
|
||
Error: "find user error: " + err.Error(),
|
||
})
|
||
return nil
|
||
}
|
||
|
||
var giftAmount int64
|
||
switch content.GiftType {
|
||
case 1:
|
||
giftAmount = int64(content.GiftValue)
|
||
case 2:
|
||
// 获取订阅对应的套餐信息
|
||
subscribeInfo, err := l.svcCtx.SubscribeModel.FindOne(context.Background(), sub.SubscribeId)
|
||
if err != nil {
|
||
*errors = append(*errors, ErrorInfo{
|
||
UserSubscribeId: sub.Id,
|
||
Error: "find subscribe error: " + err.Error(),
|
||
})
|
||
return nil
|
||
}
|
||
if subscribeInfo.UnitPrice > 0 {
|
||
giftAmount = int64(float64(subscribeInfo.UnitPrice) * (float64(content.GiftValue) / 100))
|
||
}
|
||
}
|
||
|
||
if giftAmount > 0 {
|
||
userInfo.GiftAmount += giftAmount
|
||
// 使用Update而不是Save，更精确地更新单个字段
|
||
if err := tx.Model(&user.User{}).Where("id = ?", sub.UserId).Update("gift_amount", userInfo.GiftAmount).Error; err != nil {
|
||
*errors = append(*errors, ErrorInfo{
|
||
UserSubscribeId: sub.Id,
|
||
Error: "update user gift amount error: " + err.Error(),
|
||
})
|
||
return nil
|
||
}
|
||
|
||
if err := l.createGiftLog(tx, sub.Id, userInfo.Id, giftAmount, userInfo.GiftAmount, now); err != nil {
|
||
*errors = append(*errors, ErrorInfo{
|
||
UserSubscribeId: sub.Id,
|
||
Error: "create gift log error: " + err.Error(),
|
||
})
|
||
// 回滚用户金额更新
|
||
userInfo.GiftAmount -= giftAmount
|
||
tx.Model(&user.User{}).Where("id = ?", sub.UserId).Update("gift_amount", userInfo.GiftAmount)
|
||
return nil
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (l *QuotaTaskLogic) getStartTime(sub *user.Subscribe, now time.Time) time.Time {
|
||
if sub.StartTime.Unix() == 0 {
|
||
return now
|
||
}
|
||
return sub.StartTime
|
||
}
|
||
|
||
func (l *QuotaTaskLogic) createGiftLog(tx *gorm.DB, subscribeId, userId, amount, balance int64, now time.Time) error {
|
||
giftLog := &log.Gift{
|
||
Type: log.GiftTypeIncrease,
|
||
OrderNo: "",
|
||
SubscribeId: subscribeId,
|
||
Amount: amount,
|
||
Balance: balance,
|
||
Remark: "Quota task gift",
|
||
Timestamp: now.UnixMilli(),
|
||
}
|
||
|
||
logString, err := giftLog.Marshal()
|
||
if err != nil {
|
||
return fmt.Errorf("marshal gift log error: %v", err)
|
||
}
|
||
return tx.Model(&log.SystemLog{}).Create(&log.SystemLog{
|
||
Type: log.TypeGift.Uint8(),
|
||
Content: string(logString),
|
||
ObjectID: userId,
|
||
Date: now.Format(time.DateOnly),
|
||
}).Error
|
||
}
|
||
|
||
func (l *QuotaTaskLogic) createResetTrafficLog(tx *gorm.DB, subscribeId, userId int64, now time.Time) error {
|
||
trafficLog := &log.ResetSubscribe{
|
||
Type: log.ResetSubscribeTypeQuota,
|
||
UserId: userId,
|
||
OrderNo: "",
|
||
Timestamp: now.UnixMilli(),
|
||
}
|
||
|
||
logString, err := trafficLog.Marshal()
|
||
if err != nil {
|
||
return fmt.Errorf("marshal traffic log error: %v", err)
|
||
}
|
||
return tx.Model(&log.SystemLog{}).Create(&log.SystemLog{
|
||
Type: log.TypeResetSubscribe.Uint8(),
|
||
Content: string(logString),
|
||
ObjectID: subscribeId,
|
||
Date: now.Format(time.DateOnly),
|
||
}).Error
|
||
}
|