🐛 查寝各种问题修复

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/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/controller/agent/agent_ctrlV1.go

@@ -23,7 +23,7 @@ func (a AgentCtrlV1) List(ctx context.Context, req *v1.AgentListReq) (res *utils
 // Add 添加下级代理
 func (a AgentCtrlV1) Add(ctx context.Context, req *v1.AgentAddReq) (res *utils.Response[any], err error) {
 	defer utils.Pass(&err)
-	return service.Global.Agent.Add(ctx, req), nil
+	panic("implement me")
 }
 
 // Ban 禁用代理

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/scanner.go

@@ -118,14 +118,19 @@ func (ts *TaskScanner) scanCpdailyTask() {
 			queue.AddTask(ts.ctx, asynq.NewTask(cst.TaskCpdaily, payloadWriter.Bytes()),
 				asynq.TaskID(fmt.Sprintf("cpdaily:task:%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/service/agent/agent_service.go

@@ -60,57 +60,6 @@ func (s *AgentServiceV1) List(ctx context.Context, req *v1.AgentListReq) *utils.
 	return utils.SuccessWithTotal[[]v1.AgentListRes](res, total)
 }
 
-func (s *AgentServiceV1) Add(ctx context.Context, req *v1.AgentAddReq) *utils.Response[any] {
-	// 参数校验
-	currentUser := utils.GetUserInfoFromCtx(ctx)
-	agent := currentUser.QueryAgent().OnlyX(ctx)
-	if agent.Level >= req.Level {
-		return utils.FailParam("下级代理级别不能高于当前代理级别")
-	}
-	if agent.Rate.Compare(req.Rate) <= 0 {
-		return utils.FailParam("下级代理费率比例不能低于当前代理比例")
-	}
-	tx, err := db.Global.Tx(ctx)
-	utils.OkOrPanic(err, func(err error) error {
-		return oops.
-			WithContext(ctx).
-			With("创建事务失败").
-			Wrap(err)
-	})
-	defer tx.Rollback()
-	user := tx.SysUser.Query().Where(sysuser.IDEQ(currentUser.ID)).OnlyX(ctx)
-	user.Balance = utils.CalculateBalance(user.Balance, req.Rate, req.Balance, req.Rate)
-	if user.Balance.LessThan(decimal.NewFromInt(0)) {
-		return utils.FailParam("余额不足")
-	}
-	// 扣除余额
-	_, err = tx.SysUser.UpdateOne(user).SetBalance(user.Balance).Save(ctx)
-	utils.OkOrPanic(err, func(err error) error {
-		return oops.
-			WithContext(ctx).
-			With("更新用户余额失败").
-			Wrap(err)
-	})
-	// 创建用户
-	tmp := tx.SysUser.Create().
-		SetUsername(req.Username).
-		SetPassword(utils.PasswordBcryptHash(req.Password)).
-		SetEmail(req.Email).
-		SetParent(user)
-	save := tmp.SaveX(ctx)
-	// 关联角色
-	role := tx.SysRole.Query().Where(sysrole.NameEQ(cst.KeyRoleAgent)).AllX(ctx)
-	tx.SysUser.UpdateOne(save).AddRoles(role...).ExecX(ctx)
-	// 创建代理
-	tx.SysAgent.Create().
-		SetLevel(req.Level).
-		SetRate(req.Rate).
-		SaveX(ctx)
-
-	err = tx.Commit()
-	return utils.SuccessWithMsg("")
-}
-
 func (s *AgentServiceV1) Ban(ctx context.Context, req *v1.AgentBanReq) *utils.Response[any] {
 	user := db.Global.SysUser.Query().Where(sysuser.IDEQ(req.ID)).OnlyX(ctx)
 	if !utils.IsAdminFromCtx(ctx) {
@@ -127,19 +76,22 @@ 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().WithRoles().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 {
 			return utils.FailParam("非法的下级代理参数")
 		}
-		if !utils.HasRole(currentUser.Edges.Roles, cst.KeyRoleAdmin) {
-			needBalance := utils.CalculateBalance(parent.Balance, parent.Edges.Agent.Rate, req.Amount, user.Edges.Agent.Rate)
-			balance.BalanceService.ChangeUserBalance(ctx, parent.ID, needBalance.Neg())
-		}
+		needBalance := utils.NeedBalance(parent.Edges.Agent.Rate, req.Amount, user.Edges.Agent.Rate)
+		balance.BalanceService.ChangeUserBalance(ctx, parent.ID, needBalance.Neg())
+
 	}
 	balance.BalanceService.ChangeUserBalance(ctx, user.ID, req.Amount)
 	return utils.SuccessWithMsg("操作成功")
@@ -188,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/balance/balance_service.go

@@ -37,5 +37,7 @@ func (v *BalanceServiceV1) ChangeUserBalance(ctx context.Context, userId int64,
 			Where(sysuser.ID(userId)).
 			SetBalance(newBalance).
 			ExecX(ctx)
+		tx.Commit()
 	}, nil)
+
 }

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,14 +244,21 @@ func (v CpTaskServiceV1) ExecuteNow(ctx context.Context, req *v1.ExecuteNowReq)
 
 	user := utils.GetUserInfoFromCtx(ctx)
 
-	taskExist := db.Global.CpTask.Query().
+	// 查询任务并检查状态
+	task := db.Global.CpTask.Query().
 		Where(cptask.ID(req.Id), cptask.UserID(user.ID)).
-		CountX(ctx)
+		Select(cptask.FieldTaskStatus).
+		FirstX(ctx)
 
-	if taskExist == 0 {
+	if task == nil {
 		utils.OkOrPanic(errors2.NewBusinessError(404, "任务不存在"))
 	}
 
+	// 检查任务是否正在执行中
+	if task.TaskStatus == cptask.TaskStatusDoing {
+		utils.OkOrPanic(errors2.NewBusinessError(400, "任务正在执行中，请稍后再试"))
+	}
+
 	payload := &schedule.CpdailyTaskPayload{
 		CpTaskID: req.Id,
 	}
@@ -238,22 +268,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(task.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/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()
 }

internal/utils/user.go

@@ -28,6 +28,14 @@ func CalculateBalance(balance decimal.Decimal, rate decimal.Decimal, amount deci
 	return balance.Sub(amount.Div(amountRate).Mul(rate)).Truncate(2)
 }
 
+func NeedBalance(rate decimal.Decimal, amount decimal.Decimal, amountRate decimal.Decimal) decimal.Decimal {
+	// 如果当前级别费率为0(超级管理员)
+	if rate.Equal(decimal.NewFromInt(0)) {
+		return decimal.Zero
+	}
+	return amount.Mul(rate).Div(amountRate)
+}
+
 // ConsumeBalance 消费余额
 func ConsumeBalance(balance decimal.Decimal, rate decimal.Decimal) decimal.Decimal {
 	return balance.Mul(rate)