feat: enhance task scanner functionality and improve task execution logic

internal/framework/schedule/init.go

@@ -17,8 +17,8 @@ func Init(ctx context.Context) {
 }
 
 func Stop() {
-	//globalScanner.StopScan()
-	//g.Log().Info(gctx.GetInitCtx(), "task scanner stopped")
+	globalScanner.StopScan()
+	g.Log().Info(gctx.GetInitCtx(), "task scanner stopped")
 
 	e := queue.GetClient().Close()
 	utils.OkOrPanic(e)

internal/framework/schedule/scanner.go

@@ -27,6 +27,7 @@ var globalScanner *TaskScanner
 
 func initScanner() {
 	globalScanner = NewTaskScanner()
+	go globalScanner.StartScan()
 }
 
 type TaskScanner struct {
@@ -69,7 +70,14 @@ func (ts *TaskScanner) StartScan() {
 
 func (ts *TaskScanner) StopScan() {
 	ts.cancelFunc()
-	<-ts.stopSign
+	timeoutTimer := time.NewTimer(10 * time.Second)
+	defer timeoutTimer.Stop()
+
+	select {
+	case <-timeoutTimer.C:
+		g.Log().Warningf(ts.ctx, "task scanner stop timeout")
+	case <-ts.stopSign:
+	}
 }
 
 func (ts *TaskScanner) scanCpdailyTask() {
@@ -80,7 +88,7 @@ func (ts *TaskScanner) scanCpdailyTask() {
 	minExecuteTime := currentTime.Add(-30 * time.Minute).Unix()
 
 	allTasks := db.Global.CpTask.Query().
-		Where(cptask.NextExecuteTimeLTE(currentTime.Unix()), cptask.NextExecuteTimeNotNil()).
+		Where(cptask.NextExecuteTimeLTE(currentTime.Unix()), cptask.NextExecuteTimeNotNil(), cptask.NextExecuteTimeGT(0)).
 		Select(cptask.FieldID, cptask.FieldExecuteHour, cptask.FieldExecuteMinute).
 		AllX(ts.ctx)
 

internal/service/cpdaily/cptask.go

@@ -1,20 +1,33 @@
 package cpdaily
 
 import (
+	"bytes"
 	"context"
+	"encoding/gob"
 	"encoding/json"
+	"errors"
+	"fmt"
 	v1 "likedream-goframe/internal/api/v1"
 	"likedream-goframe/internal/application/cpdaily/login"
 	"likedream-goframe/internal/application/cpdaily/school"
 	"likedream-goframe/internal/application/cpdaily/task"
 	"likedream-goframe/internal/application/cpdaily/task/formtype"
 	errors2 "likedream-goframe/internal/common/errors"
+	"likedream-goframe/internal/cst"
 	"likedream-goframe/internal/db"
 	"likedream-goframe/internal/ent"
 	"likedream-goframe/internal/ent/cptask"
 	"likedream-goframe/internal/ent/cpuser"
 	"likedream-goframe/internal/ent/sysuser"
+	"likedream-goframe/internal/framework/queue"
+	"likedream-goframe/internal/framework/schedule"
 	"likedream-goframe/internal/utils"
+	"time"
+
+	"entgo.io/ent/dialect/sql"
+	"github.com/gogf/gf/v2/frame/g"
+	"github.com/hibiken/asynq"
+	"github.com/samber/oops"
 )
 
 type CpTaskServiceV1 struct {
@@ -164,14 +177,21 @@ func (v CpTaskServiceV1) PostExecute(ctx context.Context, req *v1.PostCpTaskExec
 		utils.OkOrPanic(errors2.NewBusinessError(400, "执行次数应不低于原有次数"))
 	}
 	//todo 扣积分
-	db.Global.CpTask.UpdateOne(cpTask).
-		SetStep(cptask.StepComplete).
-		SetFrequency(req.Frequency).
-		SetTaskStatus(cptask.TaskStatusNotStart).
-		SetNotificationType(req.NotificationType).
-		SetExecuteHour(req.ExecuteHour).
-		SetExecuteMinute(req.ExecuteMinute).
-		ExecX(ctx)
+	e := db.WithTx(ctx, func(ctx context.Context, tx *ent.Tx) {
+		cpTask := db.Global.CpTask.UpdateOne(cpTask).
+			SetStep(cptask.StepComplete).
+			SetFrequency(req.Frequency).
+			SetTaskStatus(cptask.TaskStatusNotStart).
+			SetNotificationType(req.NotificationType).
+			SetExecuteHour(req.ExecuteHour).
+			SetExecuteMinute(req.ExecuteMinute).
+			SaveX(ctx)
+
+		nextExecuteAt := utils.GetCpdailyNextExecuteTime(cpTask)
+		db.Global.CpTask.UpdateOne(cpTask).SetNextExecuteTime(nextExecuteAt).ExecX(ctx)
+	}, &sql.TxOptions{})
+	utils.OkOrPanic(e)
+
 	return utils.SuccessWithMsg("更新成功")
 }
 
@@ -189,7 +209,43 @@ func (v CpTaskServiceV1) Delete(ctx context.Context, req *v1.DeleteTaskReq) *uti
 }
 
 func (v CpTaskServiceV1) ExecuteNow(ctx context.Context, req *v1.ExecuteNowReq) *utils.Response[any] {
-	// todo 立即执行
+
+	user := utils.GetUserInfoFromCtx(ctx)
+
+	taskExist := db.Global.CpTask.Query().
+		Where(cptask.ID(req.Id), cptask.UserID(user.ID)).
+		CountX(ctx)
+
+	if taskExist == 0 {
+		utils.OkOrPanic(errors2.NewBusinessError(404, "任务不存在"))
+	}
+
+	payload := &schedule.CpdailyTaskPayload{
+		CpTaskID: req.Id,
+	}
+
+	payloadWriter := bytes.NewBuffer(nil)
+
+	e := gob.NewEncoder(payloadWriter).Encode(payload)
+	utils.OkOrPanic(e, oops.WithContext(ctx).With("payload", payload).In("CpdailyTaskScanner.scanCpdailyTask").Wrap)
+
+	e = utils.Catch(func() {
+		queue.AddTask(ctx, asynq.NewTask(cst.TaskCpdaily, payloadWriter.Bytes()),
+			asynq.TaskID(fmt.Sprintf("cpdaily:task:%d", req.Id)),
+			asynq.MaxRetry(0),
+			asynq.Unique(10*time.Minute),
+			asynq.Timeout(10*time.Minute),
+			asynq.Queue(cst.TaskQueueCpdaily),
+		)
+	})
+	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, "任务执行失败"))
+		}
+	}
 	return utils.SuccessWithMsg("执行成功")
 }
 
@@ -202,13 +258,12 @@ func (v CpTaskServiceV1) PauseTask(ctx context.Context, req *v1.PauseTaskReq) *u
 		utils.OkOrPanic(errors2.NewBusinessError(400, "非法的今日校园账号"))
 	}
 	if cpTask.TaskStatus == cptask.TaskStatusPause {
-		// todo 恢复任务
-		db.Global.CpTask.UpdateOne(cpTask).SetTaskStatus(cptask.TaskStatusNotStart).ExecX(ctx)
+		nextExecuteAt := utils.GetCpdailyNextExecuteTime(cpTask)
+		db.Global.CpTask.UpdateOne(cpTask).SetTaskStatus(cptask.TaskStatusNotStart).SetNextExecuteTime(nextExecuteAt).ExecX(ctx)
 		return utils.SuccessWithMsg("恢复成功")
 
 	} else {
-		// todo 暂停任务
-		db.Global.CpTask.UpdateOne(cpTask).SetTaskStatus(cptask.TaskStatusPause).ExecX(ctx)
+		db.Global.CpTask.UpdateOne(cpTask).SetTaskStatus(cptask.TaskStatusPause).SetNextExecuteTime(-1).ExecX(ctx)
 		return utils.SuccessWithMsg("暂停成功")
 	}
 }