🐛 签到修复 & 定时重置状态

internal/application/classifier/classifier.go

@@ -45,7 +45,7 @@ func NewImageClassifier(modelPath, classesPath string) (*ImageClassifier, error)
 	ort.SetSharedLibraryPath(dir)
 	err = ort.InitializeEnvironment()
 	if err != nil {
-		logger.Logger.Info("初始化onnx：", zap.String("绝对路径", absPath), zap.String("父文件夹", dir))
+		logger.Logger.Error("初始化onnx：", zap.String("绝对路径", absPath), zap.String("父文件夹", dir))
 		return nil, fmt.Errorf("初始化ONNX Runtime失败: %v", err)
 	}
 

internal/application/cpdaily/imageuploader/upload.go

@@ -75,7 +75,7 @@ func UploadImageFromStorage(ctx context.Context, host string, imageId uuid.UUID,
 		utils.OkOrPanic(errors.NewBusinessError(503, "上传图片失败: "+utils.ReadString(resp.Body)), oops.WithContext(ctx).With("response", utils.ReadString(resp.Body)).Wrap)
 	}
 
-	previewUrl := GetPreviewUrlAsync(ctx, host, targetImage.Key, taskType, client)
+	previewUrl := GetPreviewUrlAsync(ctx, host, imageUploadPolicy.FileName, taskType, client)
 
 	return previewUrl
 }

internal/application/cpdaily/task/collection.go

@@ -161,7 +161,7 @@ func (c *CpdailyCollectionTask) QueryFormByTitle(ctx context.Context, titleGlob
 	}
 
 	if matchTask == nil {
-		utils.OkOrPanic(errors.NewBusinessError(404, fmt.Sprintf("未找到表单信息: %s", titleGlob)), func(err error) error {
+		utils.OkOrPanic(errors.NewBusinessError(503, fmt.Sprintf("未找到表单信息: %s", titleGlob)), func(err error) error {
 			return oops.With("host", c.host).
 				In("CpdailyCollectionTask.QueryFormByTitle").
 				Wrap(err)
@@ -196,7 +196,7 @@ func (c *CpdailyCollectionTask) QueryFormByTitle(ctx context.Context, titleGlob
 	detailDatas := detailJSON["datas"].MapStrVar()
 
 	if len(detailDatas) == 0 {
-		utils.OkOrPanic(errors.NewBusinessError(404, "未找到表单数据"), func(err error) error {
+		utils.OkOrPanic(errors.NewBusinessError(503, "未找到表单数据"), func(err error) error {
 			return oops.With("host", c.host).
 				In("CpdailyCollectionTask.QueryFormByTitle").
 				With("titleGlob", titleGlob).
@@ -230,7 +230,7 @@ func (c *CpdailyCollectionTask) QueryFormByTitle(ctx context.Context, titleGlob
 	formDatas := formJSON["datas"].MapStrVar()
 
 	if len(formDatas) == 0 {
-		utils.OkOrPanic(errors.NewBusinessError(404, "未找到表单数据"), func(err error) error {
+		utils.OkOrPanic(errors.NewBusinessError(503, "未找到表单数据"), func(err error) error {
 			return oops.With("host", c.host).
 				In("CpdailyCollectionTask.QueryFormByTitle").
 				With("titleGlob", titleGlob).

internal/application/cpdaily/task/sign.go

@@ -147,8 +147,8 @@ func (c *CpdailyAutoSignTask) QueryFormByTitle(ctx context.Context, titleGlob st
 
 	findMatchTitle := func(jsonArr []any) map[string]*gvar.Var {
 		for _, rawJson := range jsonArr {
-			item, ok := rawJson.(map[string]*gvar.Var)
-			if !ok {
+			item := gvar.New(rawJson).MapStrVar()
+			if len(item) == 0 {
 				continue
 			}
 			taskName, exist := item["taskName"]
@@ -178,7 +178,7 @@ func (c *CpdailyAutoSignTask) QueryFormByTitle(ctx context.Context, titleGlob st
 			if matchTask == nil {
 				matchTask = findMatchTitle(signedTasks)
 				if matchTask == nil {
-					utils.OkOrPanic(errors.NewBusinessError(404, fmt.Sprintf("任务未找到: %s", titleGlob)), func(err error) error {
+					utils.OkOrPanic(errors.NewBusinessError(503, fmt.Sprintf("任务未找到: %s", titleGlob)), func(err error) error {
 						return oops.With("host", c.host).
 							In("CpdailyAutoSignTask.QueryFormByTitle").
 							Wrap(err)
@@ -210,7 +210,7 @@ func (c *CpdailyAutoSignTask) QueryFormByTitle(ctx context.Context, titleGlob st
 	formDatas := detailResponseJSON["datas"].MapStrVar()
 
 	if len(formDatas) == 0 {
-		utils.OkOrPanic(errors.NewBusinessError(404, "任务数据为空"), func(err error) error {
+		utils.OkOrPanic(errors.NewBusinessError(503, "任务数据为空"), func(err error) error {
 			return oops.With("host", c.host).
 				In("CpdailyAutoSignTask.QueryFormByTitle").
 				With("titleGlob", titleGlob).
@@ -293,7 +293,7 @@ func (c *CpdailyAutoSignTask) QueryFormByYearMonth(ctx context.Context, titleGlo
 			if matchTask == nil {
 				matchTask = findMatchTitle(signedTasks)
 				if matchTask == nil {
-					utils.OkOrPanic(errors.NewBusinessError(404, fmt.Sprintf("任务未找到: %s", titleGlob)), func(err error) error {
+					utils.OkOrPanic(errors.NewBusinessError(503, fmt.Sprintf("任务未找到: %s", titleGlob)), func(err error) error {
 						return oops.
 							With("host", c.host).
 							With("year", year).
@@ -332,7 +332,7 @@ func (c *CpdailyAutoSignTask) QueryFormByYearMonth(ctx context.Context, titleGlo
 	formDatas := detailResponseJSON["datas"].MapStrVar()
 
 	if len(formDatas) == 0 {
-		utils.OkOrPanic(errors.NewBusinessError(404, "任务数据为空"), func(err error) error {
+		utils.OkOrPanic(errors.NewBusinessError(503, "任务数据为空"), func(err error) error {
 			return oops.
 				With("host", c.host).
 				With("year", year).
@@ -379,12 +379,12 @@ func (c *CpdailyAutoSignTask) ParseFormJson(ctx context.Context, rawJson map[str
 	dto = &formtype.CpTaskFormItem{
 		Title:    "签到具体地址",
 		ColName:  "address",
-		Type:     formtype.ItemTypeAddress,
+		Type:     formtype.ItemTypeInput,
 		Required: true,
 	}
 	list = append(list, dto)
 
-	var current uint
+	var current uint = 1
 
 	if isPhoto := rows["isPhoto"]; isPhoto != nil && isPhoto.Int() == 1 {
 		dto = &formtype.CpTaskFormItem{
@@ -539,7 +539,7 @@ func (c *CpdailyAutoSignTask) FillCustomForm(ctx context.Context, form map[strin
 
 		previewUrl := imageuploader.UploadImageFromStorage(ctx, c.host, imageId, types.TaskTypeSign, c.client)
 
-		resultForm["signImg"] = gvar.New(previewUrl)
+		resultForm["signPhotoUrl"] = gvar.New(previewUrl)
 	}
 
 	// 处理验证码

internal/application/cpdaily/task/sleepcheck.go

@@ -585,7 +585,7 @@ func (c *CpdailySleepCheckTask) FillCustomForm(ctx context.Context, form map[str
 
 		previewUrl := imageuploader.UploadImageFromStorage(ctx, c.host, imageId, types.TaskTypeSleepCheck, c.client)
 
-		resultForm["signImg"] = gvar.New(previewUrl)
+		resultForm["signPhotoUrl"] = gvar.New(previewUrl)
 	}
 
 	// 判断是否存在二维码查寝

internal/cst/task.go

@@ -1,9 +1,11 @@
 package cst
 
 const (
-	TaskCpdaily = "task:cpdaily"
-	TaskEmail   = "task:email"
+	TaskCpdaily    = "task:cpdaily"
+	TaskEmail      = "task:email"
+	TaskResetDaily = "task:reset:daily"
 
 	TaskQueueCpdaily = "queue:cpdaily"
 	TaskQueueEmail   = "queue:email"
+	TaskQueueDefault = "default"
 )

internal/ent/cptask/cptask.go

@@ -160,6 +160,7 @@ const (
 	TaskStatusDone     TaskStatus = "done"
 	TaskStatusFail     TaskStatus = "fail"
 	TaskStatusPause    TaskStatus = "pause"
+	TaskStatusQueue    TaskStatus = "queue"
 	TaskStatusUnknown  TaskStatus = "unknown"
 )
 
@@ -170,7 +171,7 @@ func (ts TaskStatus) String() string {
 // TaskStatusValidator is a validator for the "task_status" field enum values. It is called by the builders before save.
 func TaskStatusValidator(ts TaskStatus) error {
 	switch ts {
-	case TaskStatusNotStart, TaskStatusDoing, TaskStatusDone, TaskStatusFail, TaskStatusPause, TaskStatusUnknown:
+	case TaskStatusNotStart, TaskStatusDoing, TaskStatusDone, TaskStatusFail, TaskStatusPause, TaskStatusQueue, TaskStatusUnknown:
 		return nil
 	default:
 		return fmt.Errorf("cptask: invalid enum value for task_status field: %q", ts)

internal/ent/migrate/schema.go

@@ -57,7 +57,7 @@ var (
 		{Name: "task_type", Type: field.TypeEnum, Enums: []string{"collection", "sign", "sleep_check"}},
 		{Name: "title", Type: field.TypeString},
 		{Name: "frequency", Type: field.TypeInt, Default: 0},
-		{Name: "task_status", Type: field.TypeEnum, Enums: []string{"not_start", "doing", "done", "fail", "pause", "unknown"}},
+		{Name: "task_status", Type: field.TypeEnum, Enums: []string{"not_start", "doing", "done", "fail", "pause", "queue", "unknown"}},
 		{Name: "success_count", Type: field.TypeInt, Default: 0},
 		{Name: "form_data", Type: field.TypeString, Nullable: true, Size: 2147483647},
 		{Name: "step", Type: field.TypeEnum, Enums: []string{"form", "execute", "complete"}},
@@ -84,7 +84,7 @@ var (
 				Symbol:     "cp_tasks_sys_user_cp_task",
 				Columns:    []*schema.Column{CpTasksColumns[15]},
 				RefColumns: []*schema.Column{SysUserColumns[0]},
-				OnDelete:   schema.NoAction,
+				OnDelete:   schema.Cascade,
 			},
 		},
 	}
@@ -122,7 +122,7 @@ var (
 				Symbol:     "cp_task_logs_sys_user_cp_task_logs",
 				Columns:    []*schema.Column{CpTaskLogsColumns[8]},
 				RefColumns: []*schema.Column{SysUserColumns[0]},
-				OnDelete:   schema.NoAction,
+				OnDelete:   schema.Cascade,
 			},
 		},
 	}
@@ -179,7 +179,7 @@ var (
 				Symbol:     "sys_agent_sys_user_agent",
 				Columns:    []*schema.Column{SysAgentColumns[5]},
 				RefColumns: []*schema.Column{SysUserColumns[0]},
-				OnDelete:   schema.SetNull,
+				OnDelete:   schema.Cascade,
 			},
 		},
 	}

internal/ent/schema/cptask.go

@@ -22,7 +22,7 @@ func (CpTask) Fields() []ent.Field {
 		field.Enum("task_type").Values("collection", "sign", "sleep_check"),
 		field.String("title").NotEmpty(),
 		field.Int("frequency").Default(0),
-		field.Enum("task_status").Values("not_start", "doing", "done", "fail", "pause", "unknown"),
+		field.Enum("task_status").Values("not_start", "doing", "done", "fail", "pause", "queue", "unknown"),
 		field.Int("success_count").Default(0),
 		field.Text("form_data").Optional(),
 		field.Enum("step").Values("form", "execute", "complete"),

internal/framework/schedule/cpdaily_handler.go

@@ -14,6 +14,7 @@ import (
 	"likedream-goframe/internal/ent/cptask"
 	"likedream-goframe/internal/ent/cpuser"
 	"likedream-goframe/internal/framework/errors"
+	"likedream-goframe/internal/framework/logger"
 	"likedream-goframe/internal/framework/model/email"
 	email2 "likedream-goframe/internal/service/email"
 	"likedream-goframe/internal/utils"
@@ -23,6 +24,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/hibiken/asynq"
 	"github.com/samber/oops"
+	"go.uber.org/zap"
 )
 
 var emailService = email2.NewV1()
@@ -132,13 +134,15 @@ func (c *CpdailyTaskHandler) ProcessTask(ctx context.Context, task *asynq.Task)
 	submitForm := taskExecutor.FillCustomForm(ctx, cpdailyRealTask, customFormItems, deviceID, targetTask.Edges.CpUser.Username)
 	taskExecutor.Submit(ctx, submitForm, targetTask.Edges.CpUser.Username, deviceID)
 
-	db.Global.CpTask.UpdateOneID(taskPayload.CpTaskID).SetTaskStatus(cptask.TaskStatusDone).AddSuccessCount(1).ExecX(ctx)
+	// 任务完成后重置状态为 TaskStatusDone，更新下次执行时间 以便下次继续执行
+	executeTime := utils.GetCpdailyNextExecuteTime(targetTask)
+	db.Global.CpTask.UpdateOneID(taskPayload.CpTaskID).SetTaskStatus(cptask.TaskStatusDone).SetNextExecuteTime(executeTime).AddSuccessCount(1).ExecX(ctx)
 
 	// 创建日志
 	db.Global.CpTaskLog.Create().
 		SetCpTaskID(targetTask.ID).
 		SetCpUserID(targetTask.Edges.CpUser.ID).
-		SetSystemUserID(targetTask.Edges.CpUser.Edges.Sysuser.ID).
+		SetSystemUserID(targetTask.Edges.SysUser.ID).
 		SetIsSuccess(true).
 		SetRealName(realTitle).
 		ExecX(ctx)
@@ -188,5 +192,11 @@ func (c *CpdailyTaskHandler) sendTaskNotificationEmail(ctx context.Context, task
 		return // 没有邮箱地址，不发送邮件
 	}
 
-	emailService.SendTaskNotificationEmail(ctx, emailAddr, taskName, status, details)
+	// 发送邮件，捕获可能的错误并忽略，避免影响任务执行主流程
+	e := utils.Catch(func() {
+		emailService.SendTaskNotificationEmail(ctx, emailAddr, taskName, status, details)
+	})
+	if e != nil {
+		logger.Logger.Warn("发送任务通知邮件失败，已忽略", zap.Error(e))
+	}
 }

internal/framework/schedule/init.go

@@ -14,12 +14,17 @@ func Init(ctx context.Context) {
 	initServer(ctx)
 	initInspector(ctx)
 	initScanner()
+	initPeriodicTasks(ctx)
 }
 
 func Stop() {
 	globalScanner.StopScan()
 	g.Log().Info(gctx.GetInitCtx(), "task scanner stopped")
 
+	if globalPeriodicTaskManager != nil {
+		globalPeriodicTaskManager.Shutdown()
+	}
+
 	e := queue.GetClient().Close()
 	utils.OkOrPanic(e)
 	g.Log().Info(gctx.GetInitCtx(), "asynq client closed")

internal/framework/schedule/scanner.go

@@ -116,16 +116,21 @@ func (ts *TaskScanner) scanCpdailyTask() {
 
 		e = utils.Catch(func() {
 			queue.AddTask(ts.ctx, asynq.NewTask(cst.TaskCpdaily, payloadWriter.Bytes()),
-				asynq.TaskID(fmt.Sprintf("cpdaily:task:%d", task.ID)),
+				asynq.TaskID(fmt.Sprintf("cpdaily:task:schedule:%d", task.ID)),
 				asynq.MaxRetry(3),
-				asynq.Unique(1*time.Minute),
+				asynq.Unique(10*time.Second),
 				asynq.ProcessAt(execTime),
 				asynq.Deadline(execTime.Add(12*time.Minute)),
 				asynq.Queue(cst.TaskQueueCpdaily),
 			)
+			// 更新未开始的任务状态为队列中
+			db.Global.CpTask.Update().Where(cptask.ID(task.ID), cptask.TaskStatusEQ(cptask.TaskStatusNotStart)).SetTaskStatus(cptask.TaskStatusQueue).Exec(ts.ctx)
 		})
 		if e != nil {
 			if errors.Is(e, asynq.ErrDuplicateTask) {
+				// 任务已在队列中，不更新下次执行时间，等任务执行完成后再处理
+				g.Log().Debugf(ts.ctx, "任务ID %d 已在队列中，跳过本次扫描", task.ID)
+				failedIDS = append(failedIDS, task.ID)
 				continue
 			} else if errors.Is(e, context.Canceled) {
 				return
@@ -150,8 +155,9 @@ func (ts *TaskScanner) scanCpdailyTask() {
 	}
 
 	// 重置30分钟之前的任务到下次执行时间
+	// 只处理已完成配置且状态为未开始的任务，避免重复处理正在执行或已失败的任务
 	expiredTasks := db.Global.CpTask.Query().
-		Where(cptask.NextExecuteTimeLT(minExecuteTime), cptask.NextExecuteTimeNotNil(), cptask.NextExecuteTimeGT(0)).
+		Where(cptask.NextExecuteTimeLT(minExecuteTime), cptask.NextExecuteTimeNotNil(), cptask.NextExecuteTimeGT(0), cptask.StepEQ(cptask.StepComplete), cptask.TaskStatusEQ(cptask.TaskStatusNotStart)).
 		Select(cptask.FieldID, cptask.FieldExecuteHour, cptask.FieldExecuteMinute).
 		AllX(ts.ctx)
 

internal/framework/schedule/server.go

@@ -30,7 +30,8 @@ func initServer(ctx context.Context) {
 		ErrorHandler:             taskErrorHandler{},
 		Queues: map[string]int{
 			cst.TaskQueueEmail:   2,
-			cst.TaskQueueCpdaily: 8,
+			cst.TaskQueueCpdaily: 10,
+			cst.TaskQueueDefault: 1,
 		},
 	})
 
@@ -38,6 +39,7 @@ func initServer(ctx context.Context) {
 
 	mux.Handle(cst.TaskCpdaily, &CpdailyTaskHandler{})
 	mux.Handle(cst.TaskEmail, &EmailTaskHandler{})
+	mux.Handle(cst.TaskResetDaily, &ResetTaskHandler{})
 
 	e := globalServer.Start(mux)
 	utils.OkOrPanic(e)

internal/service/agent/agent_service.go

@@ -76,10 +76,14 @@ func (s *AgentServiceV1) Ban(ctx context.Context, req *v1.AgentBanReq) *utils.Re
 }
 
 func (s *AgentServiceV1) Recharge(ctx context.Context, req *v1.AgentRechargeReq) *utils.Response[any] {
-	// 参数校验
 	user := db.Global.SysUser.Query().Where(sysuser.IDEQ(req.ID)).WithAgent().OnlyX(ctx)
+
 	if !utils.IsAdminFromCtx(ctx) {
-		currentUser := db.Global.SysUser.Query().Where(sysuser.ID(utils.GetUserInfoFromCtx(ctx).ID)).WithAgent().OnlyX(ctx)
+		currentUser := utils.GetUserInfoFromCtx(ctx)
+		// 参数校验
+		if user.Edges.Agent == nil {
+			return utils.FailParam("用户不是代理")
+		}
 		// 判断是否是自己的下级代理
 		parent := user.QueryParent().WithAgent().FirstX(ctx)
 		if parent == nil || parent.ID != currentUser.ID {
@@ -136,8 +140,10 @@ func (s *AgentServiceV1) AgentPutInfoReq(ctx context.Context, req *v1.AgentPutIn
 	err = tx.Commit()
 	//按照新费率计算余额
 	newBalance := utils.TransformBalance(oldBalance, req.Rate)
-	// 6. 更新用户余额
-	balance.BalanceService.ChangeUserBalance(ctx, user.ID, newBalance.Sub(oldBalance))
+	// 6. 更新用户余额(如果更新了费率或者第一次改为代理)
+	if user.Edges.Agent == nil || !user.Edges.Agent.Rate.Equal(req.Rate) {
+		balance.BalanceService.ChangeUserBalance(ctx, user.ID, newBalance.Sub(user.Balance))
+	}
 	utils.OkOrPanic(err, func(err error) error {
 		return oops.
 			WithContext(ctx).

internal/service/cpdaily/cptask.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"encoding/gob"
 	"encoding/json"
-	"errors"
 	"fmt"
 	v1 "likedream-goframe/internal/api/v1"
 	"likedream-goframe/internal/application/cpdaily/login"
@@ -178,6 +177,10 @@ func (v CpTaskServiceV1) PostExecute(ctx context.Context, req *v1.PostCpTaskExec
 	if req.Frequency < 1 || consume < 0 {
 		utils.OkOrPanic(errors2.NewBusinessError(400, "执行次数应不低于原有次数"))
 	}
+
+	// 记录原始任务状态，用于判断是否需要立即扫描
+	originalStatus := cpTask.TaskStatus
+
 	//扣积分
 	basePoint := decimal.NewFromInt(cst.BasePointCpdaily)
 	if user.Edges.Agent != nil {
@@ -196,7 +199,27 @@ func (v CpTaskServiceV1) PostExecute(ctx context.Context, req *v1.PostCpTaskExec
 			SetExecuteMinute(req.ExecuteMinute).
 			SaveX(ctx)
 
-		nextExecuteAt := utils.GetCpdailyNextExecuteTime(cpTask)
+		// 计算下次执行时间
+		var nextExecuteAt int64
+
+		// 只有在原始状态为未开始时，才考虑5分钟窗口立即扫描
+		if originalStatus == cptask.TaskStatusNotStart {
+			currentTime := time.Now().In(cst.TimeZoneShangHai)
+			executeTime := time.Date(currentTime.Year(), currentTime.Month(), currentTime.Day(), req.ExecuteHour, req.ExecuteMinute, 0, 0, cst.TimeZoneShangHai)
+			timeDiff := executeTime.Sub(currentTime)
+
+			if timeDiff >= -5*time.Minute && timeDiff <= 5*time.Minute {
+				// 执行时间在当前时间前后5分钟内，设置为当前时间，让扫描器立即扫描到
+				nextExecuteAt = currentTime.Unix()
+			} else {
+				// 否则使用正常的计算逻辑
+				nextExecuteAt = utils.GetCpdailyNextExecuteTime(cpTask)
+			}
+		} else {
+			// 其他状态（暂停、失败等），使用正常的计算逻辑
+			nextExecuteAt = utils.GetCpdailyNextExecuteTime(cpTask)
+		}
+
 		db.Global.CpTask.UpdateOne(cpTask).SetNextExecuteTime(nextExecuteAt).ExecX(ctx)
 	}, &sql.TxOptions{})
 	utils.OkOrPanic(e)
@@ -221,13 +244,24 @@ func (v CpTaskServiceV1) ExecuteNow(ctx context.Context, req *v1.ExecuteNowReq)
 
 	user := utils.GetUserInfoFromCtx(ctx)
 
-	taskExist := db.Global.CpTask.Query().
+	// 查询任务并检查状态
+	cpTask := db.Global.CpTask.Query().
 		Where(cptask.ID(req.Id), cptask.UserID(user.ID)).
-		CountX(ctx)
+		Select(cptask.FieldTaskStatus).
+		FirstX(ctx)
 
-	if taskExist == 0 {
+	if cpTask == nil {
 		utils.OkOrPanic(errors2.NewBusinessError(404, "任务不存在"))
 	}
+	// 检查任务是否存在次数
+	if cpTask.Frequency-cpTask.SuccessCount <= 0 {
+		utils.OkOrPanic(errors2.NewBusinessError(400, "任务没有执行次数"))
+	}
+
+	// 检查任务是否正在执行中
+	if cpTask.TaskStatus != cptask.TaskStatusNotStart && cpTask.TaskStatus != cptask.TaskStatusDone && cpTask.TaskStatus != cptask.TaskStatusFail {
+		utils.OkOrPanic(errors2.NewBusinessError(400, "任务正在执行中，请稍后再试"))
+	}
 
 	payload := &schedule.CpdailyTaskPayload{
 		CpTaskID: req.Id,
@@ -238,22 +272,23 @@ func (v CpTaskServiceV1) ExecuteNow(ctx context.Context, req *v1.ExecuteNowReq)
 	e := gob.NewEncoder(payloadWriter).Encode(payload)
 	utils.OkOrPanic(e, oops.WithContext(ctx).With("payload", payload).In("CpdailyTaskScanner.scanCpdailyTask").Wrap)
 
+	// 使用时间戳生成唯一的 TaskID，避免与自动扫描的任务冲突，也避免 Unique 限制
+	// 由于已经检查了任务状态，不需要使用 Unique 来防重复
+	taskID := fmt.Sprintf("cpdaily:task:manual:%d:%d", req.Id, time.Now().UnixNano())
 	e = utils.Catch(func() {
 		queue.AddTask(ctx, asynq.NewTask(cst.TaskCpdaily, payloadWriter.Bytes()),
-			asynq.TaskID(fmt.Sprintf("cpdaily:task:%d", req.Id)),
+			asynq.TaskID(taskID),
+			asynq.Unique(5*time.Second),
 			asynq.MaxRetry(0),
-			asynq.Unique(10*time.Minute),
 			asynq.Timeout(10*time.Minute),
 			asynq.Queue(cst.TaskQueueCpdaily),
 		)
+		// 更新未开始的任务状态为队列中
+		db.Global.CpTask.Update().Where(cptask.ID(cpTask.ID), cptask.TaskStatusEQ(cptask.TaskStatusNotStart)).SetTaskStatus(cptask.TaskStatusQueue).Exec(ctx)
 	})
 	if e != nil {
-		if errors.Is(e, asynq.ErrDuplicateTask) {
-			utils.OkOrPanic(errors2.NewBusinessError(400, "任务已经在执行中"))
-		} else {
-			g.Log().Errorf(ctx, "立即执行任务入列失败: %+v", e)
-			utils.OkOrPanic(errors2.NewBusinessError(500, "任务队列中已存在此任务，立即执行任务入列失败"))
-		}
+		g.Log().Errorf(ctx, "立即执行任务入列失败: %+v", e)
+		utils.OkOrPanic(errors2.NewBusinessError(500, "同一个任务请勿连续执行"))
 	}
 	return utils.SuccessWithMsg("执行成功")
 }

internal/service/user_login/user_info_service.go

@@ -12,6 +12,7 @@ import (
 	"likedream-goframe/internal/ent/sysrole"
 	"likedream-goframe/internal/ent/sysuser"
 	"likedream-goframe/internal/framework/ent2"
+	"likedream-goframe/internal/service/balance"
 	"likedream-goframe/internal/utils"
 	"strings"
 	"time"
@@ -132,22 +133,25 @@ func (s *UserInfoServiceV1) CardRecharge(ctx context.Context, req *v1.CardRechar
 	tx, err := db.Global.Tx(ctx)
 	utils.OkOrPanic(err)
 	parent := user.QueryParent().WithAgent().FirstX(ctx)
+	// 返现20%
+	r, _ := decimal.NewFromString("0.2")
+	cashbackValue := decimal.NewFromUint64(card.Value).Mul(r)
 	if parent != nil {
-		// 返现20%
-		r, _ := decimal.NewFromString("0.2")
-		cashbackValue := decimal.NewFromUint64(card.Value).Mul(r)
 		// 这里查询父级是否是代理，代理的话要x代理费率
 		if parent.Edges.Agent != nil {
 			cashbackValue = cashbackValue.Mul(parent.Edges.Agent.Rate)
 		}
-		tx.SysUser.UpdateOneID(user.Edges.Parent.ID).SetBalance(parent.Balance.Add(cashbackValue)).SetInviteCashback(parent.InviteCashback.Add(cashbackValue)).ExecX(ctx)
+		tx.SysUser.UpdateOneID(user.Edges.Parent.ID).SetInviteCashback(parent.InviteCashback.Add(cashbackValue)).ExecX(ctx)
 		tx.SysCard.UpdateOneID(card.ID).SetStatus(syscard.StatusCashback).SetUsedID(user.ID).SetCashUserID(user.Edges.Parent.ID).SetCashValue(cashbackValue).ExecX(ctx)
 	} else {
 		//tx.SysCard.UpdateOneID(card.ID).SetStatus(cst.CardStatusUsed).SetUsedID(user.ID).ExecX(ctx)
 		tx.SysCard.UpdateOneID(card.ID).SetStatus(syscard.StatusUsed).SetUsedID(user.ID).ExecX(ctx)
 	}
-	tx.SysUser.UpdateOneID(user.ID).SetBalance(user.Balance.Add(decimal.NewFromUint64(card.Value))).ExecX(ctx)
 	err = tx.Commit()
+	balance.BalanceService.ChangeUserBalance(ctx, user.ID, decimal.NewFromUint64(card.Value))
+	if parent != nil {
+		balance.BalanceService.ChangeUserBalance(ctx, parent.ID, cashbackValue)
+	}
 	utils.OkOrPanic(err, func(err error) error {
 		return oops.
 			WithContext(ctx).

internal/utils/schedule.go

@@ -26,7 +26,7 @@ func GetCpdailyNextExecuteTime(task *ent.CpTask) int64 {
 
 	// 如果当天执行时间已过，返回明天的执行时间
 	executeTime := time.Date(current.Year(), current.Month(), current.Day(), *executeHour, *executeMinute, 0, 0, cst.TimeZoneShangHai)
-	executeTime.AddDate(0, 0, 1)
+	executeTime = executeTime.AddDate(0, 0, 1)
 
 	return executeTime.Unix()
 }