feat: worker 支持通过 cron 触发

Author: YangZxi

.gitignore

@@ -11,3 +11,6 @@ dist/
 
 # Go
 coverage.out
+
+# other
+.work_note.md

.vscode/settings.json

@@ -0,0 +1,2 @@
+{
+}

AGENTS.md

@@ -24,15 +24,13 @@
 
 ```text
 cmd/main.go                 # 程序入口，启动 Router/Admin 双服务
-internal/admin              # Admin API（鉴权、Worker 管理、文件管理）
-internal/router             # Router 动态执行入口
-internal/executor           # 脚本执行与超时控制
-internal/requestparse       # JSON/form/multipart 解析
-internal/registry           # 路由内存索引
-internal/db                 # SQLite 持久化
-internal/model              # 领域模型（包含 FuncInput/FuncRequest/FuncOutput）
-pages/src                   # Admin 服务的前端源码（Settings/Workers/WorkerInfo/Dependencies）
-public                      # 前端构建产物（由后端直接托管）
+internal/admin              # admin 服务，负责项目所有管理功能
+internal/router             # worker 的 http 路由定义
+internal/executor           # http、cron 触发调用 worker 时的执行器
+internal/common             # 公共通用模块，JSON/form/multipart 解析
+internal/db                 # 数据库层，使用 ORM 提供操作数据的接口
+internal/model              # 数据库实体与 Go 实体
+pages/                      # 前端源代码
 migrations/001_init.sql     # 建表 SQL
 .github/workflows           # CI（镜像构建推送）
 ```
@@ -138,6 +136,7 @@ docker compose up --build
 3. 关键流程允许添加简洁中文注释，避免注释噪音。
 4. 新增路径、包名、模型名时，必须全局搜索引用并一次性更新。
 5. 禁止使用无意义命名和难以理解的随意缩写；允许使用行业内约定俗成、语义明确的常见缩写（如 `cfg`、`reg`、`dao`），但所有变量名、函数名、方法名都必须名如其意，看到名称就能理解其职责和用途。
+6. 避免做不需要的兜底开发，一般情况下，你只需要考虑当下就行，不必考虑到未来的各种参数值不合法问题。在不明确时，询问我是否需要进行兜底。
 
 ## 8. 变更流程（代理执行）
 

cmd/main.go

@@ -16,8 +16,9 @@ import (
 
 	"callit/internal/admin"
 	"callit/internal/config"
+	"callit/internal/cron"
 	"callit/internal/db"
-	"callit/internal/registry"
+	"callit/internal/executor"
 	"callit/internal/router"
 
 	"golang.org/x/sync/errgroup"
@@ -38,15 +39,21 @@ func main() {
 		log.Fatalf("加载应用配置失败: %v", err)
 	}
 
-	reg := registry.New()
+	reg := router.New()
 	funcs, err := store.Worker.ListEnabled(context.Background())
 	if err != nil {
 		log.Fatalf("加载启用函数失败: %v", err)
 	}
 	reg.Reload(funcs)
 
-	routerEngine := router.NewEngine(store, reg, cfg.DataDir)
-	adminEngine := admin.NewEngine(store, reg, cfg)
+	invoker := executor.NewService(store, cfg.DataDir)
+	cronManager := cron.NewManager(store, invoker, time.Local)
+	if err := cronManager.Start(context.Background()); err != nil {
+		log.Fatalf("启动 cron 调度器失败: %v", err)
+	}
+
+	routerEngine := router.NewEngine(store, reg, cfg.DataDir, invoker)
+	adminEngine := admin.NewEngine(store, reg, cronManager, cfg)
 	handler := serverRouteHandler(adminEngine, routerEngine, cfg.AdminPrefix)
 
 	srv := &http.Server{

internal/admin/message/request.go

@@ -54,3 +54,19 @@ type ChatStreamRequest struct {
 	Message      string `json:"message"`
 	HistoryLimit int    `json:"history_limit"`
 }
+
+// CreateCronTaskRequest 表示创建 cron_task 的请求体。
+type CreateCronTaskRequest struct {
+	Cron string `json:"cron"`
+}
+
+// UpdateCronTaskRequest 表示更新 cron_task 的请求体。
+type UpdateCronTaskRequest struct {
+	ID   int64  `json:"id,string"`
+	Cron string `json:"cron"`
+}
+
+// CronTaskIDRequest 表示仅包含 cron_task 主键 ID 的请求体。
+type CronTaskIDRequest struct {
+	ID int64 `json:"id,string"`
+}

internal/admin/server.go

@@ -22,10 +22,12 @@ import (
 	"callit/internal/admin/chat"
 	"callit/internal/admin/message"
 	"callit/internal/common"
+	"callit/internal/common/snowflake"
 	"callit/internal/config"
+	"callit/internal/cron"
 	"callit/internal/db"
 	"callit/internal/model"
-	"callit/internal/registry"
+	"callit/internal/router"
 
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
@@ -41,12 +43,14 @@ const (
 
 // Server 表示 Admin 服务。
 type Server struct {
-	store       *db.Store
-	reg         *registry.Registry
-	dataDir     string
-	adminToken  string
-	chatHandler *chat.Handler
-	configMu    sync.RWMutex
+	store        *db.Store
+	reg          *router.Registry
+	cronReloader interface{ Reload(context.Context) error }
+	idGenerator  *snowflake.Generator
+	dataDir      string
+	adminToken   string
+	chatHandler  *chat.Handler
+	configMu     sync.RWMutex
 
 	dependencyTaskMu      sync.Mutex
 	dependencyTaskRunning bool
@@ -69,13 +73,15 @@ func apiError(c *gin.Context, httpStatus int, msg string) {
 }
 
 // NewEngine 创建 Admin Gin 引擎。
-func NewEngine(store *db.Store, reg *registry.Registry, cfg config.Config) *gin.Engine {
+func NewEngine(store *db.Store, reg *router.Registry, cronReloader interface{ Reload(context.Context) error }, cfg config.Config) *gin.Engine {
 	s := &Server{
-		store:       store,
-		reg:         reg,
-		dataDir:     cfg.DataDir,
-		adminToken:  cfg.AdminToken,
-		chatHandler: chat.NewHandler(store, cfg.DataDir, cfg.AppConfig),
+		store:        store,
+		reg:          reg,
+		cronReloader: cronReloader,
+		idGenerator:  snowflake.NewGenerator(1),
+		dataDir:      cfg.DataDir,
+		adminToken:   cfg.AdminToken,
+		chatHandler:  chat.NewHandler(store, cfg.DataDir, cfg.AppConfig),
 	}
 	e := gin.New()
 	e.Use(gin.Recovery(), common.RequestIDMiddleware())
@@ -102,6 +108,10 @@ func NewEngine(store *db.Store, reg *registry.Registry, cfg config.Config) *gin.
 		api.POST("/workers/disable", s.disableWorker)
 
 		api.GET("/workers/:id/logs", s.listWorkerLogs)
+		api.GET("/workers/:id/crons", s.listWorkerCrons)
+		api.POST("/workers/:id/crons/create", s.createWorkerCron)
+		api.POST("/workers/:id/crons/update", s.updateWorkerCron)
+		api.POST("/workers/:id/crons/delete", s.deleteWorkerCron)
 
 		api.GET("/workers/:id/files", s.listWorkerFiles)
 		api.GET("/workers/:id/files/:filename", s.getFileContent)
@@ -266,7 +276,7 @@ func (s *Server) createWorker(c *gin.Context) {
 		return
 	}
 
-	if err := s.reloadRegistry(c.Request.Context()); err != nil {
+	if err := s.reloadWorkersState(c.Request.Context()); err != nil {
 		apiError(c, http.StatusInternalServerError, err.Error())
 		return
 	}
@@ -343,7 +353,7 @@ func (s *Server) updateWorker(c *gin.Context) {
 		return
 	}
 
-	if err := s.reloadRegistry(c.Request.Context()); err != nil {
+	if err := s.reloadWorkersState(c.Request.Context()); err != nil {
 		apiError(c, http.StatusInternalServerError, err.Error())
 		return
 	}
@@ -451,13 +461,17 @@ func (s *Server) deleteWorker(c *gin.Context) {
 		apiError(c, http.StatusInternalServerError, err.Error())
 		return
 	}
+	if err := s.store.CronTask.DeleteByWorkerID(c.Request.Context(), id); err != nil {
+		apiError(c, http.StatusInternalServerError, err.Error())
+		return
+	}
 
 	functionDir := filepath.Join(s.dataDir, "workers", id)
 	if err := os.RemoveAll(functionDir); err != nil {
 		apiError(c, http.StatusInternalServerError, fmt.Sprintf("删除函数目录失败: %v", err))
 		return
 	}
-	if err := s.reloadRegistry(c.Request.Context()); err != nil {
+	if err := s.reloadWorkersState(c.Request.Context()); err != nil {
 		apiError(c, http.StatusInternalServerError, err.Error())
 		return
 	}
@@ -807,13 +821,25 @@ func (s *Server) setWorkerEnabled(c *gin.Context, enabled bool) {
 		apiError(c, http.StatusInternalServerError, err.Error())
 		return
 	}
-	if err := s.reloadRegistry(c.Request.Context()); err != nil {
+	if err := s.reloadWorkersState(c.Request.Context()); err != nil {
 		apiError(c, http.StatusInternalServerError, err.Error())
 		return
 	}
 	apiSuccess(c, updated)
 }
 
+func (s *Server) reloadWorkersState(ctx context.Context) error {
+	if err := s.reloadRegistry(ctx); err != nil {
+		return err
+	}
+	if s.cronReloader != nil {
+		if err := s.cronReloader.Reload(ctx); err != nil {
+			return fmt.Errorf("重载 cron 调度器失败: %w", err)
+		}
+	}
+	return nil
+}
+
 func (s *Server) reloadRegistry(ctx context.Context) error {
 	reloadCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
 	defer cancel()
@@ -825,6 +851,157 @@ func (s *Server) reloadRegistry(ctx context.Context) error {
 	return nil
 }
 
+func (s *Server) listWorkerCrons(c *gin.Context) {
+	workerID, ok := s.requireWorker(c)
+	if !ok {
+		return
+	}
+
+	tasks, err := s.store.CronTask.ListByWorkerID(c.Request.Context(), workerID)
+	if err != nil {
+		apiError(c, http.StatusInternalServerError, err.Error())
+		return
+	}
+	apiSuccess(c, tasks)
+}
+
+func (s *Server) createWorkerCron(c *gin.Context) {
+	workerID, ok := s.requireWorker(c)
+	if !ok {
+		return
+	}
+
+	var req message.CreateCronTaskRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		apiError(c, http.StatusBadRequest, "请求体格式错误")
+		return
+	}
+
+	cronExpr := strings.TrimSpace(req.Cron)
+	if err := s.validateCronExpression(cronExpr); err != nil {
+		apiError(c, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	taskID, err := s.idGenerator.NextID()
+	if err != nil {
+		apiError(c, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	created, err := s.store.CronTask.Create(c.Request.Context(), model.CronTask{
+		ID:       taskID,
+		WorkerID: workerID,
+		Cron:     cronExpr,
+	})
+	if err != nil {
+		apiError(c, http.StatusInternalServerError, err.Error())
+		return
+	}
+	if err := s.reloadWorkersState(c.Request.Context()); err != nil {
+		apiError(c, http.StatusInternalServerError, err.Error())
+		return
+	}
+	apiSuccess(c, created)
+}
+
+func (s *Server) updateWorkerCron(c *gin.Context) {
+	workerID, ok := s.requireWorker(c)
+	if !ok {
+		return
+	}
+
+	var req message.UpdateCronTaskRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		apiError(c, http.StatusBadRequest, "请求体格式错误")
+		return
+	}
+	if req.ID <= 0 {
+		apiError(c, http.StatusBadRequest, "id 不能为空")
+		return
+	}
+
+	cronExpr := strings.TrimSpace(req.Cron)
+	if err := s.validateCronExpression(cronExpr); err != nil {
+		apiError(c, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	updated, err := s.store.CronTask.Update(c.Request.Context(), model.CronTask{
+		ID:       req.ID,
+		WorkerID: workerID,
+		Cron:     cronExpr,
+	})
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			apiError(c, http.StatusNotFound, "cron_task 不存在")
+			return
+		}
+		apiError(c, http.StatusInternalServerError, err.Error())
+		return
+	}
+	if err := s.reloadWorkersState(c.Request.Context()); err != nil {
+		apiError(c, http.StatusInternalServerError, err.Error())
+		return
+	}
+	apiSuccess(c, updated)
+}
+
+func (s *Server) deleteWorkerCron(c *gin.Context) {
+	workerID, ok := s.requireWorker(c)
+	if !ok {
+		return
+	}
+
+	var req message.CronTaskIDRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		apiError(c, http.StatusBadRequest, "请求体格式错误")
+		return
+	}
+	if req.ID <= 0 {
+		apiError(c, http.StatusBadRequest, "id 不能为空")
+		return
+	}
+
+	if err := s.store.CronTask.Delete(c.Request.Context(), req.ID, workerID); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			apiError(c, http.StatusNotFound, "cron_task 不存在")
+			return
+		}
+		apiError(c, http.StatusInternalServerError, err.Error())
+		return
+	}
+	if err := s.reloadWorkersState(c.Request.Context()); err != nil {
+		apiError(c, http.StatusInternalServerError, err.Error())
+		return
+	}
+	apiSuccess(c, gin.H{"ok": true})
+}
+
+func (s *Server) requireWorker(c *gin.Context) (string, bool) {
+	workerID := strings.TrimSpace(c.Param("id"))
+	if workerID == "" {
+		apiError(c, http.StatusBadRequest, "id 不能为空")
+		return "", false
+	}
+	if _, err := s.store.Worker.GetByID(c.Request.Context(), workerID); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			apiError(c, http.StatusNotFound, "函数不存在")
+			return "", false
+		}
+		apiError(c, http.StatusInternalServerError, err.Error())
+		return "", false
+	}
+	return workerID, true
+}
+
+func (s *Server) validateCronExpression(expr string) error {
+	if strings.TrimSpace(expr) == "" {
+		return fmt.Errorf("cron 不能为空")
+	}
+	return cron.ValidateExpression(expr)
+}
+
 func validateWorker(worker model.Worker) error {
 	if strings.TrimSpace(worker.Name) == "" {
 		return fmt.Errorf("name 不能为空")