feat(dashboard): cleanup tasks and events (#2368)

assets

@@ -1 +1 @@
-Subproject commit 6a6fd722f35d3fca9eb1c34297e3f3eae8c9a161
+Subproject commit e9b91c4e03654d5968f8a676a13fc4badf530b5d

inventory/task.go

@@ -3,6 +3,7 @@ package inventory
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"entgo.io/ent/dialect/sql"
 	"github.com/cloudreve/Cloudreve/v4/ent"
@@ -44,6 +45,8 @@ type TaskClient interface {
 	List(ctx context.Context, args *ListTaskArgs) (*ListTaskResult, error)
 	// DeleteByIDs deletes the tasks with the given IDs.
 	DeleteByIDs(ctx context.Context, ids ...int) error
+	// DeleteBy deletes the tasks with the given args.
+	DeleteBy(ctx context.Context, args *DeleteTaskArgs) error
 }
 
 type (
@@ -59,6 +62,12 @@ type (
 		*PaginationResults
 		Tasks []*ent.Task
 	}
+
+	DeleteTaskArgs struct {
+		NotAfter time.Time
+		Types    []string
+		Status   []task.Status
+	}
 )
 
 func NewTaskClient(client *ent.Client, dbType conf.DBType, hasher hashid.Encoder) TaskClient {
@@ -113,6 +122,23 @@ func (c *taskClient) DeleteByIDs(ctx context.Context, ids ...int) error {
 	return err
 }
 
+func (c *taskClient) DeleteBy(ctx context.Context, args *DeleteTaskArgs) error {
+	query := c.client.Task.
+		Delete().
+		Where(task.CreatedAtLTE(args.NotAfter))
+
+	if len(args.Status) > 0 {
+		query.Where(task.StatusIn(args.Status...))
+	}
+
+	if len(args.Types) > 0 {
+		query.Where(task.TypeIn(args.Types...))
+	}
+
+	_, err := query.Exec(ctx)
+	return err
+}
+
 func (c *taskClient) Update(ctx context.Context, task *ent.Task, args *TaskArgs) (*ent.Task, error) {
 	stm := c.client.Task.UpdateOne(task).
 		SetPublicState(args.PublicState)

routers/controllers/admin.go

@@ -518,6 +518,17 @@ func AdminBatchDeleteEntity(c *gin.Context) {
 	}
 }
 
+func AdminCleanupTask(c *gin.Context) {
+	service := ParametersFromContext[*admin.CleanupTaskService](c, admin.CleanupTaskParameterCtx{})
+	err := service.CleanupTask(c)
+	if err != nil {
+		c.JSON(200, serializer.Err(c, err))
+		return
+	}
+
+	c.JSON(200, serializer.Response{})
+}
+
 func AdminListTasks(c *gin.Context) {
 	service := ParametersFromContext[*admin.AdminListService](c, admin.AdminListServiceParamsCtx{})
 	res, err := service.Tasks(c)

routers/router.go

@@ -854,6 +854,11 @@ func initMasterRouter(dep dependency.Dep) *gin.Engine {
 						controllers.FromJSON[adminsvc.BatchTaskService](adminsvc.BatchTaskParamCtx{}),
 						controllers.AdminBatchDeleteTask,
 					)
+					// Cleanup tasks
+					queue.POST("cleanup",
+						controllers.FromJSON[adminsvc.CleanupTaskService](adminsvc.CleanupTaskParameterCtx{}),
+						controllers.AdminCleanupTask,
+					)
 					// // 列出任务
 					// queue.POST("list", controllers.AdminListTask)
 					// // 新建文件导入任务

service/admin/task.go

@@ -3,6 +3,7 @@ package admin
 import (
 	"context"
 	"strconv"
+	"time"
 
 	"github.com/cloudreve/Cloudreve/v4/application/dependency"
 	"github.com/cloudreve/Cloudreve/v4/ent"
@@ -251,3 +252,31 @@ func (s *BatchTaskService) Delete(c *gin.Context) error {
 
 	return nil
 }
+
+type (
+	CleanupTaskService struct {
+		NotAfter time.Time     `json:"not_after" binding:"required"`
+		Types    []string      `json:"types"`
+		Status   []task.Status `json:"status"`
+	}
+	CleanupTaskParameterCtx struct{}
+)
+
+func (s *CleanupTaskService) CleanupTask(c *gin.Context) error {
+	dep := dependency.FromContext(c)
+	taskClient := dep.TaskClient()
+
+	if len(s.Status) == 0 {
+		s.Status = []task.Status{task.StatusCanceled, task.StatusCompleted, task.StatusError}
+	}
+
+	if err := taskClient.DeleteBy(c, &inventory.DeleteTaskArgs{
+		NotAfter: s.NotAfter,
+		Types:    s.Types,
+		Status:   s.Status,
+	}); err != nil {
+		return serializer.NewError(serializer.CodeDBError, "Failed to cleanup tasks", err)
+	}
+
+	return nil
+}