diff --git a/.gitignore b/.gitignore index 950fb61..e1d9728 100644 --- a/.gitignore +++ b/.gitignore @@ -18,6 +18,7 @@ backend/config.yaml # 4. 临时文件与日志 (Logs & Temp) *.log /tmp/ +/.tmp/ /frontend/dist/ /scripts/ diff --git a/backend/cmd/start.go b/backend/cmd/start.go index a22b197..b626cdc 100644 --- a/backend/cmd/start.go +++ b/backend/cmd/start.go @@ -18,6 +18,7 @@ import ( gatewaynotification "github.com/LoveLosita/smartflow/backend/gateway/client/notification" gatewayschedule "github.com/LoveLosita/smartflow/backend/gateway/client/schedule" gatewaytask "github.com/LoveLosita/smartflow/backend/gateway/client/task" + gatewaytaskclass "github.com/LoveLosita/smartflow/backend/gateway/client/taskclass" gatewayuserauth "github.com/LoveLosita/smartflow/backend/gateway/client/userauth" gatewayrouter "github.com/LoveLosita/smartflow/backend/gateway/router" kafkabus "github.com/LoveLosita/smartflow/backend/infra/kafka" @@ -239,6 +240,14 @@ func buildRuntime(ctx context.Context) (*appRuntime, error) { if err != nil { return nil, fmt.Errorf("failed to initialize task zrpc client: %w", err) } + taskClassClient, err := gatewaytaskclass.NewClient(gatewaytaskclass.ClientConfig{ + Endpoints: viper.GetStringSlice("taskClass.rpc.endpoints"), + Target: viper.GetString("taskClass.rpc.target"), + Timeout: viper.GetDuration("taskClass.rpc.timeout"), + }) + if err != nil { + return nil, fmt.Errorf("failed to initialize task-class zrpc client: %w", err) + } activeSchedulerClient, err := gatewayactivescheduler.NewClient(gatewayactivescheduler.ClientConfig{ Endpoints: viper.GetStringSlice("activeScheduler.rpc.endpoints"), Target: viper.GetString("activeScheduler.rpc.target"), @@ -254,7 +263,6 @@ func buildRuntime(ctx context.Context) (*appRuntime, error) { taskSv := service.NewTaskService(taskRepo, cacheRepo, taskOutboxPublisher) taskSv.SetActiveScheduleDAO(manager.ActiveSchedule) courseService := buildCourseService(llmService, courseRepo, scheduleRepo) - taskClassService := service.NewTaskClassService(taskClassRepo, cacheRepo, scheduleRepo, manager) scheduleService := service.NewScheduleService(scheduleRepo, taskClassRepo, manager, cacheRepo) agentService := service.NewAgentServiceWithSchedule( llmService, @@ -317,7 +325,7 @@ func buildRuntime(ctx context.Context) (*appRuntime, error) { return nil, err } agentService.SetActiveScheduleSessionRerunFunc(buildActiveScheduleSessionRerunFunc(manager.ActiveSchedule, activeScheduleGraphRunner, activeSchedulePreviewConfirm, activeScheduleFeedbackLocator)) - handlers := buildAPIHandlers(taskClient, taskClassService, courseService, scheduleClient, agentService, memoryModule, activeSchedulerClient, notificationClient) + handlers := buildAPIHandlers(taskClient, taskClassClient, courseService, scheduleClient, agentService, memoryModule, activeSchedulerClient, notificationClient) runtime := &appRuntime{ db: db, @@ -906,7 +914,7 @@ func buildQuickTaskQueryFunc(agentService *service.AgentService) func(ctx contex func buildAPIHandlers( taskClient ports.TaskCommandClient, - taskClassService *service.TaskClassService, + taskClassClient ports.TaskClassCommandClient, courseService *service.CourseService, scheduleClient ports.ScheduleCommandClient, agentService *service.AgentService, @@ -916,7 +924,7 @@ func buildAPIHandlers( ) *api.ApiHandlers { return &api.ApiHandlers{ TaskHandler: api.NewTaskHandler(taskClient), - TaskClassHandler: api.NewTaskClassHandler(taskClassService), + TaskClassHandler: api.NewTaskClassHandler(taskClassClient), CourseHandler: api.NewCourseHandler(courseService), ScheduleHandler: api.NewScheduleAPI(scheduleClient), AgentHandler: api.NewAgentHandler(agentService), diff --git a/backend/cmd/task-class/main.go b/backend/cmd/task-class/main.go new file mode 100644 index 0000000..3981318 --- /dev/null +++ b/backend/cmd/task-class/main.go @@ -0,0 +1,66 @@ +package main + +import ( + "context" + "log" + "os" + "os/signal" + "syscall" + + "github.com/LoveLosita/smartflow/backend/bootstrap" + rootdao "github.com/LoveLosita/smartflow/backend/dao" + rootmiddleware "github.com/LoveLosita/smartflow/backend/middleware" + taskclassdao "github.com/LoveLosita/smartflow/backend/services/task_class/dao" + taskclassrpc "github.com/LoveLosita/smartflow/backend/services/task_class/rpc" + taskclasssv "github.com/LoveLosita/smartflow/backend/services/task_class/sv" + "github.com/spf13/viper" +) + +func main() { + if err := bootstrap.LoadConfig(); err != nil { + log.Fatalf("failed to load config: %v", err) + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + db, err := taskclassdao.OpenDBFromConfig() + if err != nil { + log.Fatalf("failed to connect task-class database: %v", err) + } + redisClient, err := taskclassdao.OpenRedisFromConfig() + if err != nil { + log.Fatalf("failed to connect task-class redis: %v", err) + } + defer redisClient.Close() + + cacheRepo := rootdao.NewCacheDAO(redisClient) + if err := db.Use(rootmiddleware.NewGormCachePlugin(cacheRepo)); err != nil { + log.Fatalf("failed to initialize task-class cache deleter: %v", err) + } + + // 1. task-class 自有 DAO 使用新服务目录下的实现，保持所有权边界清晰。 + // 2. scheduleRepo / RepoManager 属于迁移期桥接依赖，用来保留原 insert/apply 的本地事务语义。 + taskClassRepo := taskclassdao.NewTaskClassDAO(db) + scheduleRepo := rootdao.NewScheduleDAO(db) + manager := rootdao.NewManager(db) + svc := taskclasssv.NewTaskClassService(taskClassRepo, cacheRepo, scheduleRepo, manager) + + server, listenOn, err := taskclassrpc.NewServer(taskclassrpc.ServerOptions{ + ListenOn: viper.GetString("taskClass.rpc.listenOn"), + Timeout: viper.GetDuration("taskClass.rpc.timeout"), + Service: svc, + }) + if err != nil { + log.Fatalf("failed to build task-class zrpc server: %v", err) + } + defer server.Stop() + + go func() { + log.Printf("task-class zrpc service starting on %s", listenOn) + server.Start() + }() + + <-ctx.Done() + log.Println("task-class service stopping") +} diff --git a/backend/config.example.yaml b/backend/config.example.yaml index b8eb63d..e2ba76d 100644 --- a/backend/config.example.yaml +++ b/backend/config.example.yaml @@ -75,6 +75,14 @@ task: - "127.0.0.1:9085" timeout: 6s +# 任务类服务配置。 +taskClass: + rpc: + listenOn: "0.0.0.0:9086" + endpoints: + - "127.0.0.1:9086" + timeout: 6s + # 主动调度服务配置。 activeScheduler: rpc: diff --git a/backend/dao/schedule.go b/backend/dao/schedule.go index cdb07ec..9e6bf97 100644 --- a/backend/dao/schedule.go +++ b/backend/dao/schedule.go @@ -418,7 +418,9 @@ func (d *ScheduleDAO) SetScheduleEmbeddedTaskIDToNull(ctx context.Context, event } func (d *ScheduleDAO) FindEmbeddedTaskIDAndDeleteIt(ctx context.Context, taskID int) (int, error) { - // 1. 先找到 schedules 表中 embedded_task_id = taskID 的记录，获取对应的 event_id + // 1. 先找到 schedules 表中 embedded_task_id = taskID 的记录，获取对应的 event_id。 + // 1.1 该 taskID 可能是“嵌入课程”的任务块，也可能是“独立任务日程”的任务块； + // 1.2 两者撤销策略不同：课程只清 embedded_task_id，独立任务需要删除 schedules 后再删 event。 type row struct { EventID *int `gorm:"column:event_id"` } @@ -438,30 +440,72 @@ func (d *ScheduleDAO) FindEmbeddedTaskIDAndDeleteIt(ctx context.Context, taskID } eventID := *r.EventID - // 2. 删除该 event_id 对应的课程事件（通过级联删除实现） - res := d.db.WithContext(ctx). - Table("schedule_events"). + var event model.ScheduleEvent + if err := d.db.WithContext(ctx). Where("id = ?", eventID). - Delete(&model.ScheduleEvent{}) - if res.Error != nil { - return 0, res.Error + First(&event).Error; err != nil { + return 0, err } - if res.RowsAffected == 0 { + + if event.Type == "task" && event.RelID != nil && *event.RelID == taskID { + // 2. 独立任务日程：schedules.event_id 是外键，必须先删原子槽位再删事件。 + if err := d.db.WithContext(ctx). + Table("schedules"). + Where("event_id = ?", eventID). + Delete(&model.Schedule{}).Error; err != nil { + return 0, err + } + res := d.db.WithContext(ctx). + Table("schedule_events"). + Where("id = ?", eventID). + Delete(&model.ScheduleEvent{}) + if res.Error != nil { + return 0, res.Error + } + if res.RowsAffected == 0 { + return 0, respond.TargetTaskNotEmbeddedInAnySchedule + } + return eventID, nil + } + + // 3. 嵌入课程：保留课程事件与课程槽位，只清空 embedded_task_id。 + clearRes := d.db.WithContext(ctx). + Table("schedules"). + Where("embedded_task_id = ?", taskID). + Update("embedded_task_id", nil) + if clearRes.Error != nil { + return 0, clearRes.Error + } + if clearRes.RowsAffected == 0 { return 0, respond.TargetTaskNotEmbeddedInAnySchedule } return eventID, nil } func (d *ScheduleDAO) DeleteScheduleEventByTaskItemID(ctx context.Context, taskItemID int) error { - //直接找schedule_events表中type=task且rel_id=taskItemID的记录，删除它（级联删schedules） - res := d.db.WithContext(ctx). + // 1. 先找 type=task 且 rel_id=taskItemID 的正式事件；若前一步已经删除则保持幂等成功。 + var eventIDs []int + if err := d.db.WithContext(ctx). Table("schedule_events"). Where("type = ? AND rel_id = ?", "task", taskItemID). - Delete(&model.ScheduleEvent{}) - if res.Error != nil { - return res.Error + Pluck("id", &eventIDs).Error; err != nil { + return err } - return nil + if len(eventIDs) == 0 { + return nil + } + + // 2. schedules.event_id 指向 schedule_events.id，删除顺序必须先子表后父表。 + if err := d.db.WithContext(ctx). + Table("schedules"). + Where("event_id IN ?", eventIDs). + Delete(&model.Schedule{}).Error; err != nil { + return err + } + return d.db.WithContext(ctx). + Table("schedule_events"). + Where("id IN ?", eventIDs). + Delete(&model.ScheduleEvent{}).Error } func (d *ScheduleDAO) GetUserRecentCompletedSchedules(ctx context.Context, nowTime time.Time, userID int, index, limit int) ([]model.Schedule, error) { diff --git a/backend/gateway/api/task-class.go b/backend/gateway/api/task-class.go index 0d65aca..60ae9f4 100644 --- a/backend/gateway/api/task-class.go +++ b/backend/gateway/api/task-class.go @@ -6,41 +6,34 @@ import ( "strconv" "time" - "github.com/LoveLosita/smartflow/backend/model" "github.com/LoveLosita/smartflow/backend/respond" - "github.com/LoveLosita/smartflow/backend/service" + taskclasscontracts "github.com/LoveLosita/smartflow/backend/shared/contracts/taskclass" + "github.com/LoveLosita/smartflow/backend/shared/ports" "github.com/gin-gonic/gin" ) +const taskClassRequestTimeout = 6 * time.Second + type TaskClassHandler struct { - svc *service.TaskClassService + client ports.TaskClassCommandClient } -// NewTaskClassHandler 组装 Handler 的“工厂” -func NewTaskClassHandler(svc *service.TaskClassService) *TaskClassHandler { - return &TaskClassHandler{ - svc: svc, // 把传进来的 Service 揣进口袋里 - } +// NewTaskClassHandler 创建 task-class HTTP 门面。 +func NewTaskClassHandler(client ports.TaskClassCommandClient) *TaskClassHandler { + return &TaskClassHandler{client: client} } -const ( - create = 0 - update = 1 -) - func (api *TaskClassHandler) UserAddTaskClass(c *gin.Context) { - var req model.UserAddTaskClassRequest - err := c.ShouldBindJSON(&req) - if err != nil { + var req taskclasscontracts.UpsertTaskClassRequest + if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, respond.WrongParamType) return } - userIDInterface := c.GetInt("user_id") - // 创建一个带 1 秒超时的上下文 - ctx, cancel := context.WithTimeout(c.Request.Context(), 1*time.Second) - defer cancel() // 记得释放资源 - err = api.svc.AddOrUpdateTaskClass(ctx, &req, userIDInterface, create, 0) - if err != nil { + req.UserID = c.GetInt("user_id") + + ctx, cancel := context.WithTimeout(c.Request.Context(), taskClassRequestTimeout) + defer cancel() + if _, err := api.client.AddTaskClass(ctx, req); err != nil { respond.DealWithError(c, err) return } @@ -48,11 +41,11 @@ func (api *TaskClassHandler) UserAddTaskClass(c *gin.Context) { } func (api *TaskClassHandler) UserGetTaskClassInfos(c *gin.Context) { - userIDInterface := c.GetInt("user_id") - // 创建一个带 1 秒超时的上下文 - ctx, cancel := context.WithTimeout(c.Request.Context(), 1*time.Second) - defer cancel() // 记得释放资源 - resp, err := api.svc.GetUserTaskClassInfos(ctx, userIDInterface) + userID := c.GetInt("user_id") + + ctx, cancel := context.WithTimeout(c.Request.Context(), taskClassRequestTimeout) + defer cancel() + resp, err := api.client.ListTaskClasses(ctx, userID) if err != nil { respond.DealWithError(c, err) return @@ -62,7 +55,6 @@ func (api *TaskClassHandler) UserGetTaskClassInfos(c *gin.Context) { func (api *TaskClassHandler) UserGetCompleteTaskClass(c *gin.Context) { taskClassID := c.Query("task_class_id") - //将taskClassID转换为int intTaskClassID, err := strconv.Atoi(taskClassID) if err != nil { c.JSON(http.StatusBadRequest, respond.WrongParamType) @@ -72,11 +64,14 @@ func (api *TaskClassHandler) UserGetCompleteTaskClass(c *gin.Context) { c.JSON(http.StatusBadRequest, respond.MissingParam) return } - userIDInterface := c.GetInt("user_id") - // 创建一个带 1 秒超时的上下文 - ctx, cancel := context.WithTimeout(c.Request.Context(), 1*time.Second) - defer cancel() // 记得释放资源 - resp, err := api.svc.GetUserCompleteTaskClass(ctx, userIDInterface, intTaskClassID) + userID := c.GetInt("user_id") + + ctx, cancel := context.WithTimeout(c.Request.Context(), taskClassRequestTimeout) + defer cancel() + resp, err := api.client.GetTaskClass(ctx, taskclasscontracts.GetTaskClassRequest{ + UserID: userID, + TaskClassID: intTaskClassID, + }) if err != nil { respond.DealWithError(c, err) return @@ -85,21 +80,23 @@ func (api *TaskClassHandler) UserGetCompleteTaskClass(c *gin.Context) { } func (api *TaskClassHandler) UserUpdateTaskClass(c *gin.Context) { - var req model.UserAddTaskClassRequest - err := c.ShouldBindJSON(&req) - if err != nil { + var req taskclasscontracts.UpsertTaskClassRequest + if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, respond.WrongParamType) return } taskClassID := c.Query("task_class_id") - //将taskClassID转换为int intTaskClassID, err := strconv.Atoi(taskClassID) - userIDInterface := c.GetInt("user_id") - // 创建一个带 1 秒超时的上下文 - ctx, cancel := context.WithTimeout(c.Request.Context(), 1*time.Second) - defer cancel() // 记得释放资源 - err = api.svc.AddOrUpdateTaskClass(ctx, &req, userIDInterface, update, intTaskClassID) if err != nil { + c.JSON(http.StatusBadRequest, respond.WrongParamType) + return + } + req.UserID = c.GetInt("user_id") + req.TaskClassID = intTaskClassID + + ctx, cancel := context.WithTimeout(c.Request.Context(), taskClassRequestTimeout) + defer cancel() + if _, err := api.client.UpdateTaskClass(ctx, req); err != nil { respond.DealWithError(c, err) return } @@ -107,25 +104,23 @@ func (api *TaskClassHandler) UserUpdateTaskClass(c *gin.Context) { } func (api *TaskClassHandler) UserAddTaskClassItemIntoSchedule(c *gin.Context) { - var req model.UserInsertTaskClassItemToScheduleRequest - err := c.ShouldBindJSON(&req) - if err != nil { + var req taskclasscontracts.InsertTaskClassItemIntoScheduleRequest + if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, respond.WrongParamType) return } taskID := c.Query("task_item_id") - //将taskID转换为int intTaskID, err := strconv.Atoi(taskID) if err != nil { c.JSON(http.StatusBadRequest, respond.WrongParamType) return } - userIDInterface := c.GetInt("user_id") - // 创建一个带 1 秒超时的上下文 - ctx, cancel := context.WithTimeout(c.Request.Context(), 1*time.Second) - defer cancel() // 记得释放资源 - err = api.svc.AddTaskClassItemIntoSchedule(ctx, &req, userIDInterface, intTaskID) - if err != nil { + req.UserID = c.GetInt("user_id") + req.TaskItemID = intTaskID + + ctx, cancel := context.WithTimeout(c.Request.Context(), taskClassRequestTimeout) + defer cancel() + if _, err := api.client.InsertTaskClassItemIntoSchedule(ctx, req); err != nil { respond.DealWithError(c, err) return } @@ -134,18 +129,19 @@ func (api *TaskClassHandler) UserAddTaskClassItemIntoSchedule(c *gin.Context) { func (api *TaskClassHandler) DeleteTaskClassItem(c *gin.Context) { taskID := c.Query("task_item_id") - //将taskID转换为int intTaskID, err := strconv.Atoi(taskID) if err != nil { c.JSON(http.StatusBadRequest, respond.WrongParamType) return } userID := c.GetInt("user_id") - // 创建一个带 1 秒超时的上下文 - ctx, cancel := context.WithTimeout(c.Request.Context(), 1*time.Second) - defer cancel() // 记得释放资源 - err = api.svc.DeleteTaskClassItem(ctx, userID, intTaskID) - if err != nil { + + ctx, cancel := context.WithTimeout(c.Request.Context(), taskClassRequestTimeout) + defer cancel() + if _, err := api.client.DeleteTaskClassItem(ctx, taskclasscontracts.DeleteTaskClassItemRequest{ + UserID: userID, + TaskItemID: intTaskID, + }); err != nil { respond.DealWithError(c, err) return } @@ -154,18 +150,19 @@ func (api *TaskClassHandler) DeleteTaskClassItem(c *gin.Context) { func (api *TaskClassHandler) DeleteTaskClass(c *gin.Context) { taskClassID := c.Query("task_class_id") - //将taskClassID转换为int intTaskClassID, err := strconv.Atoi(taskClassID) if err != nil { c.JSON(http.StatusBadRequest, respond.WrongParamType) return } userID := c.GetInt("user_id") - // 创建一个带 1 秒超时的上下文 - ctx, cancel := context.WithTimeout(c.Request.Context(), 1*time.Second) - defer cancel() // 记得释放资源 - err = api.svc.DeleteTaskClass(ctx, userID, intTaskClassID) - if err != nil { + + ctx, cancel := context.WithTimeout(c.Request.Context(), taskClassRequestTimeout) + defer cancel() + if _, err := api.client.DeleteTaskClass(ctx, taskclasscontracts.DeleteTaskClassRequest{ + UserID: userID, + TaskClassID: intTaskClassID, + }); err != nil { respond.DealWithError(c, err) return } @@ -173,18 +170,16 @@ func (api *TaskClassHandler) DeleteTaskClass(c *gin.Context) { } func (api *TaskClassHandler) UserInsertBatchTaskClassItemsIntoSchedule(c *gin.Context) { - var req model.UserInsertTaskClassItemToScheduleRequestBatch - err := c.ShouldBindJSON(&req) - if err != nil { + var req taskclasscontracts.ApplyBatchIntoScheduleRequest + if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, respond.WrongParamType) return } - userID := c.GetInt("user_id") - // 创建一个带 1 秒超时的上下文 - ctx, cancel := context.WithTimeout(c.Request.Context(), 1*time.Second) - defer cancel() // 记得释放资源 - err = api.svc.BatchApplyPlans(ctx, req.TaskClassID, userID, &req) - if err != nil { + req.UserID = c.GetInt("user_id") + + ctx, cancel := context.WithTimeout(c.Request.Context(), taskClassRequestTimeout) + defer cancel() + if _, err := api.client.ApplyBatchIntoSchedule(ctx, req); err != nil { respond.DealWithError(c, err) return } diff --git a/backend/gateway/client/taskclass/client.go b/backend/gateway/client/taskclass/client.go new file mode 100644 index 0000000..2234b4f --- /dev/null +++ b/backend/gateway/client/taskclass/client.go @@ -0,0 +1,154 @@ +package taskclass + +import ( + "context" + "encoding/json" + "errors" + "strings" + "time" + + taskclasspb "github.com/LoveLosita/smartflow/backend/services/task_class/rpc/pb" + taskclasscontracts "github.com/LoveLosita/smartflow/backend/shared/contracts/taskclass" + "github.com/zeromicro/go-zero/zrpc" + "google.golang.org/grpc" +) + +const ( + defaultEndpoint = "127.0.0.1:9086" + defaultTimeout = 6 * time.Second +) + +type ClientConfig struct { + Endpoints []string + Target string + Timeout time.Duration +} + +// Client 是 gateway 访问 task-class zrpc 的最小适配层。 +// +// 职责边界： +// 1. 只负责跨进程 gRPC 调用和 JSON 透传，不触碰 DAO 或迁移期 schedule 直写细节； +// 2. HTTP 入参仍由 gateway/api 做基础绑定，业务校验交给 task-class 服务； +// 3. 复杂响应不在 gateway 重建模型，避免 DTO 复制扩散。 +type Client struct { + rpc taskclasspb.TaskClassClient +} + +func NewClient(cfg ClientConfig) (*Client, error) { + timeout := cfg.Timeout + if timeout <= 0 { + timeout = defaultTimeout + } + endpoints := normalizeEndpoints(cfg.Endpoints) + target := strings.TrimSpace(cfg.Target) + if len(endpoints) == 0 && target == "" { + endpoints = []string{defaultEndpoint} + } + + zclient, err := zrpc.NewClient(zrpc.RpcClientConf{ + Endpoints: endpoints, + Target: target, + NonBlock: true, + Timeout: int64(timeout / time.Millisecond), + }) + if err != nil { + return nil, err + } + client := &Client{rpc: taskclasspb.NewTaskClassClient(zclient.Conn())} + if err := client.ping(timeout); err != nil { + return nil, err + } + return client, nil +} + +func (c *Client) AddTaskClass(ctx context.Context, req taskclasscontracts.UpsertTaskClassRequest) (json.RawMessage, error) { + resp, err := c.callJSON(ctx, c.rpc.AddTaskClass, req) + return jsonFromResponse(resp, err) +} + +func (c *Client) ListTaskClasses(ctx context.Context, userID int) (json.RawMessage, error) { + resp, err := c.callJSON(ctx, c.rpc.ListTaskClasses, taskclasscontracts.UserRequest{UserID: userID}) + return jsonFromResponse(resp, err) +} + +func (c *Client) GetTaskClass(ctx context.Context, req taskclasscontracts.GetTaskClassRequest) (json.RawMessage, error) { + resp, err := c.callJSON(ctx, c.rpc.GetTaskClass, req) + return jsonFromResponse(resp, err) +} + +func (c *Client) UpdateTaskClass(ctx context.Context, req taskclasscontracts.UpsertTaskClassRequest) (json.RawMessage, error) { + resp, err := c.callJSON(ctx, c.rpc.UpdateTaskClass, req) + return jsonFromResponse(resp, err) +} + +func (c *Client) InsertTaskClassItemIntoSchedule(ctx context.Context, req taskclasscontracts.InsertTaskClassItemIntoScheduleRequest) (json.RawMessage, error) { + resp, err := c.callJSON(ctx, c.rpc.InsertTaskClassItemIntoSchedule, req) + return jsonFromResponse(resp, err) +} + +func (c *Client) DeleteTaskClassItem(ctx context.Context, req taskclasscontracts.DeleteTaskClassItemRequest) (json.RawMessage, error) { + resp, err := c.callJSON(ctx, c.rpc.DeleteTaskClassItem, req) + return jsonFromResponse(resp, err) +} + +func (c *Client) DeleteTaskClass(ctx context.Context, req taskclasscontracts.DeleteTaskClassRequest) (json.RawMessage, error) { + resp, err := c.callJSON(ctx, c.rpc.DeleteTaskClass, req) + return jsonFromResponse(resp, err) +} + +func (c *Client) ApplyBatchIntoSchedule(ctx context.Context, req taskclasscontracts.ApplyBatchIntoScheduleRequest) (json.RawMessage, error) { + resp, err := c.callJSON(ctx, c.rpc.ApplyBatchIntoSchedule, req) + return jsonFromResponse(resp, err) +} + +func (c *Client) ensureReady() error { + if c == nil || c.rpc == nil { + return errors.New("task-class zrpc client is not initialized") + } + return nil +} + +func (c *Client) ping(timeout time.Duration) error { + if err := c.ensureReady(); err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + _, err := c.rpc.Ping(ctx, &taskclasspb.StatusResponse{}) + return responseFromRPCError(err) +} + +func (c *Client) callJSON(ctx context.Context, fn func(context.Context, *taskclasspb.JSONRequest, ...grpc.CallOption) (*taskclasspb.JSONResponse, error), payload any) (*taskclasspb.JSONResponse, error) { + if err := c.ensureReady(); err != nil { + return nil, err + } + raw, err := json.Marshal(payload) + if err != nil { + return nil, err + } + return fn(ctx, &taskclasspb.JSONRequest{PayloadJson: raw}) +} + +func jsonFromResponse(resp *taskclasspb.JSONResponse, rpcErr error) (json.RawMessage, error) { + if rpcErr != nil { + return nil, responseFromRPCError(rpcErr) + } + if resp == nil { + return nil, errors.New("task-class zrpc service returned empty JSON response") + } + if len(resp.DataJson) == 0 { + return json.RawMessage("null"), nil + } + return json.RawMessage(resp.DataJson), nil +} + +func normalizeEndpoints(values []string) []string { + endpoints := make([]string, 0, len(values)) + for _, value := range values { + trimmed := strings.TrimSpace(value) + if trimmed != "" { + endpoints = append(endpoints, trimmed) + } + } + return endpoints +} diff --git a/backend/gateway/client/taskclass/errors.go b/backend/gateway/client/taskclass/errors.go new file mode 100644 index 0000000..80f631e --- /dev/null +++ b/backend/gateway/client/taskclass/errors.go @@ -0,0 +1,96 @@ +package taskclass + +import ( + "errors" + "fmt" + "strings" + + "github.com/LoveLosita/smartflow/backend/respond" + "google.golang.org/genproto/googleapis/rpc/errdetails" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// responseFromRPCError 负责把 task-class 的 gRPC 错误反解回项目内错误。 +// +// 职责边界： +// 1. 只在 gateway 边缘层使用； +// 2. 业务错误尽量恢复成 respond.Response，方便 API 层复用 DealWithError； +// 3. 服务不可用或未知内部错误包装成普通 error，避免误报成用户可修正的参数问题。 +func responseFromRPCError(err error) error { + if err == nil { + return nil + } + + st, ok := status.FromError(err) + if !ok { + return wrapRPCError(err) + } + if resp, ok := responseFromStatus(st); ok { + return resp + } + + switch st.Code() { + case codes.Internal, codes.Unknown, codes.Unavailable, codes.DeadlineExceeded, codes.DataLoss, codes.Unimplemented: + msg := strings.TrimSpace(st.Message()) + if msg == "" { + msg = "task-class zrpc service internal error" + } + return wrapRPCError(errors.New(msg)) + } + + msg := strings.TrimSpace(st.Message()) + if msg == "" { + msg = "task-class zrpc service rejected request" + } + return respond.Response{Status: grpcCodeToRespondStatus(st.Code()), Info: msg} +} + +func responseFromStatus(st *status.Status) (respond.Response, bool) { + if st == nil { + return respond.Response{}, false + } + for _, detail := range st.Details() { + info, ok := detail.(*errdetails.ErrorInfo) + if !ok { + continue + } + statusValue := strings.TrimSpace(info.Reason) + if statusValue == "" { + statusValue = grpcCodeToRespondStatus(st.Code()) + } + message := strings.TrimSpace(st.Message()) + if message == "" && info.Metadata != nil { + message = strings.TrimSpace(info.Metadata["info"]) + } + if message == "" { + message = statusValue + } + return respond.Response{Status: statusValue, Info: message}, true + } + return respond.Response{}, false +} + +func grpcCodeToRespondStatus(code codes.Code) string { + switch code { + case codes.Unauthenticated: + return respond.ErrUnauthorized.Status + case codes.PermissionDenied: + return respond.ErrUnauthorized.Status + case codes.InvalidArgument: + return respond.MissingParam.Status + case codes.NotFound: + return respond.UserTaskClassNotFound.Status + case codes.Internal, codes.Unknown, codes.DataLoss: + return "500" + default: + return "400" + } +} + +func wrapRPCError(err error) error { + if err == nil { + return nil + } + return fmt.Errorf("调用 task-class zrpc 服务失败: %w", err) +} diff --git a/backend/services/schedule/dao/schedule.go b/backend/services/schedule/dao/schedule.go index cdb07ec..9e6bf97 100644 --- a/backend/services/schedule/dao/schedule.go +++ b/backend/services/schedule/dao/schedule.go @@ -418,7 +418,9 @@ func (d *ScheduleDAO) SetScheduleEmbeddedTaskIDToNull(ctx context.Context, event } func (d *ScheduleDAO) FindEmbeddedTaskIDAndDeleteIt(ctx context.Context, taskID int) (int, error) { - // 1. 先找到 schedules 表中 embedded_task_id = taskID 的记录，获取对应的 event_id + // 1. 先找到 schedules 表中 embedded_task_id = taskID 的记录，获取对应的 event_id。 + // 1.1 该 taskID 可能是“嵌入课程”的任务块，也可能是“独立任务日程”的任务块； + // 1.2 两者撤销策略不同：课程只清 embedded_task_id，独立任务需要删除 schedules 后再删 event。 type row struct { EventID *int `gorm:"column:event_id"` } @@ -438,30 +440,72 @@ func (d *ScheduleDAO) FindEmbeddedTaskIDAndDeleteIt(ctx context.Context, taskID } eventID := *r.EventID - // 2. 删除该 event_id 对应的课程事件（通过级联删除实现） - res := d.db.WithContext(ctx). - Table("schedule_events"). + var event model.ScheduleEvent + if err := d.db.WithContext(ctx). Where("id = ?", eventID). - Delete(&model.ScheduleEvent{}) - if res.Error != nil { - return 0, res.Error + First(&event).Error; err != nil { + return 0, err } - if res.RowsAffected == 0 { + + if event.Type == "task" && event.RelID != nil && *event.RelID == taskID { + // 2. 独立任务日程：schedules.event_id 是外键，必须先删原子槽位再删事件。 + if err := d.db.WithContext(ctx). + Table("schedules"). + Where("event_id = ?", eventID). + Delete(&model.Schedule{}).Error; err != nil { + return 0, err + } + res := d.db.WithContext(ctx). + Table("schedule_events"). + Where("id = ?", eventID). + Delete(&model.ScheduleEvent{}) + if res.Error != nil { + return 0, res.Error + } + if res.RowsAffected == 0 { + return 0, respond.TargetTaskNotEmbeddedInAnySchedule + } + return eventID, nil + } + + // 3. 嵌入课程：保留课程事件与课程槽位，只清空 embedded_task_id。 + clearRes := d.db.WithContext(ctx). + Table("schedules"). + Where("embedded_task_id = ?", taskID). + Update("embedded_task_id", nil) + if clearRes.Error != nil { + return 0, clearRes.Error + } + if clearRes.RowsAffected == 0 { return 0, respond.TargetTaskNotEmbeddedInAnySchedule } return eventID, nil } func (d *ScheduleDAO) DeleteScheduleEventByTaskItemID(ctx context.Context, taskItemID int) error { - //直接找schedule_events表中type=task且rel_id=taskItemID的记录，删除它（级联删schedules） - res := d.db.WithContext(ctx). + // 1. 先找 type=task 且 rel_id=taskItemID 的正式事件；若前一步已经删除则保持幂等成功。 + var eventIDs []int + if err := d.db.WithContext(ctx). Table("schedule_events"). Where("type = ? AND rel_id = ?", "task", taskItemID). - Delete(&model.ScheduleEvent{}) - if res.Error != nil { - return res.Error + Pluck("id", &eventIDs).Error; err != nil { + return err } - return nil + if len(eventIDs) == 0 { + return nil + } + + // 2. schedules.event_id 指向 schedule_events.id，删除顺序必须先子表后父表。 + if err := d.db.WithContext(ctx). + Table("schedules"). + Where("event_id IN ?", eventIDs). + Delete(&model.Schedule{}).Error; err != nil { + return err + } + return d.db.WithContext(ctx). + Table("schedule_events"). + Where("id IN ?", eventIDs). + Delete(&model.ScheduleEvent{}).Error } func (d *ScheduleDAO) GetUserRecentCompletedSchedules(ctx context.Context, nowTime time.Time, userID int, index, limit int) ([]model.Schedule, error) { diff --git a/backend/services/task_class/dao/connect.go b/backend/services/task_class/dao/connect.go new file mode 100644 index 0000000..64d2e65 --- /dev/null +++ b/backend/services/task_class/dao/connect.go @@ -0,0 +1,76 @@ +package dao + +import ( + "context" + "fmt" + + "github.com/LoveLosita/smartflow/backend/model" + "github.com/go-redis/redis/v8" + "github.com/spf13/viper" + "gorm.io/driver/mysql" + "gorm.io/gorm" +) + +// OpenDBFromConfig 创建 task-class 服务自己的数据库句柄。 +// +// 职责边界： +// 1. 只迁移 task_classes / task_items 这两个 task-class 自有表； +// 2. 不迁移 schedule_events / schedules，迁移期只检查它们是否存在； +// 3. 迁移期允许 task-class 继续直写 schedule 表，以保留原本本地事务语义。 +func OpenDBFromConfig() (*gorm.DB, error) { + host := viper.GetString("database.host") + port := viper.GetString("database.port") + user := viper.GetString("database.user") + password := viper.GetString("database.password") + dbname := viper.GetString("database.dbname") + + dsn := fmt.Sprintf( + "%s:%s@tcp(%s:%s)/%s?charset=utf8mb4&parseTime=True&loc=Local", + user, password, host, port, dbname, + ) + + db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{}) + if err != nil { + return nil, err + } + if err = db.AutoMigrate(&model.TaskClass{}, &model.TaskClassItem{}); err != nil { + return nil, fmt.Errorf("auto migrate task-class tables failed: %w", err) + } + if err = ensureRuntimeDependencyTables(db); err != nil { + return nil, err + } + return db, nil +} + +// OpenRedisFromConfig 创建 task-class 服务自己的 Redis 句柄。 +// +// 职责边界： +// 1. 只负责初始化 task-class 列表缓存和幂等链路所需的 Redis client； +// 2. 不清理任何业务 key； +// 3. Ping 失败直接返回错误，避免服务启动后才暴露缓存不可用。 +func OpenRedisFromConfig() (*redis.Client, error) { + client := redis.NewClient(&redis.Options{ + Addr: viper.GetString("redis.host") + ":" + viper.GetString("redis.port"), + Password: viper.GetString("redis.password"), + DB: 0, + }) + if _, err := client.Ping(context.Background()).Result(); err != nil { + return nil, err + } + return client, nil +} + +// ensureRuntimeDependencyTables 显式检查 task-class 迁移期仍直写的外部表。 +// +// 说明： +// 1. schedule_events / schedules 属于 schedule 服务正式日程域； +// 2. 本轮按主人拍板保留 task-class 直写权限，换取 insert/apply 与 item 状态更新的本地事务语义； +// 3. 后续若改为 schedule RPC bridge，应先补幂等与补偿，再从这里移除依赖检查。 +func ensureRuntimeDependencyTables(db *gorm.DB) error { + for _, table := range []string{"schedule_events", "schedules"} { + if !db.Migrator().HasTable(table) { + return fmt.Errorf("task-class runtime dependency table missing: %s", table) + } + } + return nil +} diff --git a/backend/services/task_class/dao/task_class.go b/backend/services/task_class/dao/task_class.go new file mode 100644 index 0000000..dd8fe17 --- /dev/null +++ b/backend/services/task_class/dao/task_class.go @@ -0,0 +1,350 @@ +package dao + +import ( + "context" + "errors" + + "github.com/LoveLosita/smartflow/backend/model" + "github.com/LoveLosita/smartflow/backend/respond" + "gorm.io/gorm" +) + +type TaskClassDAO struct { + // 这是一个口袋，用来装数据库连接实例 + db *gorm.DB +} + +// NewTaskClassDAO 创建TaskClassDAO实例 +// NewTaskClassDAO 接收一个 *gorm.DB，并把它塞进结构体的口袋里 +func NewTaskClassDAO(db *gorm.DB) *TaskClassDAO { + return &TaskClassDAO{ + db: db, + } +} + +func (dao *TaskClassDAO) WithTx(tx *gorm.DB) *TaskClassDAO { + return &TaskClassDAO{ + db: tx, + } +} + +// AddOrUpdateTaskClass 为指定用户添加/更新任务类（防越权：更新时限定 user_id） +func (dao *TaskClassDAO) AddOrUpdateTaskClass(userID int, taskClass *model.TaskClass) (int, error) { + // 不信任入参里的 UserID，强制使用当前登录用户 + taskClass.UserID = &userID + + // 新增：ID == 0 直接插入 + if taskClass.ID == 0 { + if err := dao.db.Create(taskClass).Error; err != nil { + return 0, err + } + return taskClass.ID, nil + } + // 更新：必须同时匹配 id + user_id，否则不会更新任何行（避免覆盖他人数据） + tx := dao.db.Model(&model.TaskClass{}). + Where("id = ? AND user_id = ?", taskClass.ID, userID). + Updates(taskClass) + if tx.Error != nil { + return 0, tx.Error + } + if tx.RowsAffected == 0 { + // 未匹配到记录：要么不存在，要么不属于该用户 + return 0, respond.UserTaskClassForbidden + } + return taskClass.ID, nil +} + +func (dao *TaskClassDAO) AddOrUpdateTaskClassItems(userID int, items []model.TaskClassItem) error { + if len(items) == 0 { + return nil + } + + // 1) 校验这些 items 关联的 task_class（category_id）都属于当前用户 + categoryIDSet := make(map[int]struct{}, len(items)) + var categoryIDs []int + for _, it := range items { + if *it.CategoryID == 0 { + return gorm.ErrRecordNotFound + } + if _, ok := categoryIDSet[*it.CategoryID]; !ok { + categoryIDSet[*it.CategoryID] = struct{}{} + categoryIDs = append(categoryIDs, *it.CategoryID) + } + } + + var count int64 + if err := dao.db.Model(&model.TaskClass{}). + Where("id IN ? AND user_id = ?", categoryIDs, userID). + Count(&count).Error; err != nil { + return err + } + if count != int64(len(categoryIDs)) { + return respond.UserTaskClassForbidden + } + + // 2) 新增与更新分开处理：新增不受影响；更新时限定 category_id（防越权） + var toCreate []model.TaskClassItem + for _, it := range items { + if it.ID == 0 { + toCreate = append(toCreate, it) + continue + } + + tx := dao.db.Model(&model.TaskClassItem{}). + Where("id = ? AND category_id IN ?", it.ID, categoryIDs). + Updates(map[string]any{ + "category_id": it.CategoryID, + }) + if tx.Error != nil { + return tx.Error + } + if tx.RowsAffected == 0 { + return respond.UserTaskClassForbidden + } + } + + if len(toCreate) > 0 { + if err := dao.db.Create(&toCreate).Error; err != nil { + return err + } + } + return nil +} + +// Transaction 在一个事务中执行传入的函数，供 service 层复用（自动提交/回滚） +// 规则：fn 返回 nil -> commit；fn 返回 error 或 panic -> rollback +func (dao *TaskClassDAO) Transaction(fn func(txDAO *TaskClassDAO) error) error { + return dao.db.Transaction(func(tx *gorm.DB) error { + return fn(NewTaskClassDAO(tx)) + }) +} + +func (dao *TaskClassDAO) GetUserTaskClasses(userID int) ([]model.TaskClass, error) { + var taskClasses []model.TaskClass + err := dao.db.Where("user_id = ?", userID).Find(&taskClasses).Error + if err != nil { + return nil, err + } + return taskClasses, nil +} + +// GetCompleteTaskClassByID 带着 ID 和 UserID 去取，防越权 +func (dao *TaskClassDAO) GetCompleteTaskClassByID(ctx context.Context, id int, userID int) (*model.TaskClass, error) { + var taskClass model.TaskClass + + // 1. 使用 Preload("Items") 自动执行两条 SQL 并组装 + // SQL A: SELECT * FROM task_classes WHERE id = ? AND user_id = ? + // SQL B: SELECT * FROM task_class_items WHERE category_id = (SQL A 的 ID) + err := dao.db.WithContext(ctx). + Preload("Items"). + Where("id = ? AND user_id = ?", id, userID). + First(&taskClass).Error + + if err != nil { + return nil, err + } + return &taskClass, nil +} + +// GetCompleteTaskClassesByIDs 批量获取“完整任务类”（含 Items）。 +// +// 职责边界： +// 1. 负责按 user_id + ids 过滤，保证数据归属安全； +// 2. 负责预加载 Items，供智能粗排直接使用； +// 3. 不负责排序策略，返回结果顺序由 service 层决定； +// 4. 若存在任一 id 不存在或不属于该用户，返回 WrongTaskClassID。 +func (dao *TaskClassDAO) GetCompleteTaskClassesByIDs(ctx context.Context, userID int, ids []int) ([]model.TaskClass, error) { + if len(ids) == 0 { + return []model.TaskClass{}, nil + } + + // 1. 先做去重与合法值过滤，避免无效 ID 放大数据库压力。 + uniqueIDs := make([]int, 0, len(ids)) + seen := make(map[int]struct{}, len(ids)) + for _, id := range ids { + if id <= 0 { + continue + } + if _, exists := seen[id]; exists { + continue + } + seen[id] = struct{}{} + uniqueIDs = append(uniqueIDs, id) + } + if len(uniqueIDs) == 0 { + return nil, respond.WrongTaskClassID + } + + // 2. 批量查询并预加载任务项。 + var taskClasses []model.TaskClass + err := dao.db.WithContext(ctx). + Preload("Items"). + Where("user_id = ? AND id IN ?", userID, uniqueIDs). + Find(&taskClasses).Error + if err != nil { + return nil, err + } + + // 3. 数量校验：少一条都视为“存在非法/越权 ID”，统一按业务错误返回。 + if len(taskClasses) != len(uniqueIDs) { + return nil, respond.WrongTaskClassID + } + return taskClasses, nil +} + +func (dao *TaskClassDAO) GetTaskClassItemByID(ctx context.Context, id int) (*model.TaskClassItem, error) { + var item model.TaskClassItem + err := dao.db.WithContext(ctx). + Where("id = ?", id). + First(&item).Error + if err != nil { + return nil, err + } + return &item, nil +} + +func (dao *TaskClassDAO) GetTaskClassIDByTaskItemID(ctx context.Context, itemID int) (int, error) { + var item model.TaskClassItem + res := dao.db.WithContext(ctx). + Select("category_id"). + Where("id = ?", itemID). + First(&item) + if res.Error != nil { + if errors.Is(res.Error, gorm.ErrRecordNotFound) { + return 0, respond.TaskClassItemNotFound + } + return 0, res.Error + } + return *item.CategoryID, nil +} + +func (dao *TaskClassDAO) GetTaskClassUserIDByID(ctx context.Context, taskClassID int) (int, error) { + var taskClass model.TaskClass + err := dao.db.WithContext(ctx). + Select("user_id"). + Where("id = ?", taskClassID). + First(&taskClass).Error + if err != nil { + return 0, err + } + return *taskClass.UserID, nil +} + +func (dao *TaskClassDAO) UpdateTaskClassItemEmbeddedTime(ctx context.Context, taskID int, embeddedTime *model.TargetTime) error { + err := dao.db.WithContext(ctx). + Model(&model.TaskClassItem{}). + Where("id = ?", taskID). + Update("embedded_time", embeddedTime).Error + return err +} + +func (dao *TaskClassDAO) DeleteTaskClassItemEmbeddedTime(ctx context.Context, taskID int) error { + err := dao.db.WithContext(ctx). + Model(&model.TaskClassItem{}). + Where("id = ?", taskID). + Update("embedded_time", nil).Error + return err +} + +func (dao *TaskClassDAO) IfTaskClassItemArranged(ctx context.Context, taskID int) (bool, error) { + var item model.TaskClassItem + err := dao.db.WithContext(ctx). + Select("embedded_time"). + Where("id = ?", taskID). + First(&item).Error + if err != nil { + return false, err + } + return item.EmbeddedTime != nil, nil +} + +func (dao *TaskClassDAO) BatchCheckIfTaskClassItemsArranged(ctx context.Context, itemIDs []int) (bool, error) { + if len(itemIDs) == 0 { + return false, nil + } + var count int64 + err := dao.db.WithContext(ctx). + Model(&model.TaskClassItem{}). + Where("id IN ? AND embedded_time IS NOT NULL", itemIDs). + Count(&count).Error + if err != nil { + return false, err + } + return count > 0, nil +} + +func (dao *TaskClassDAO) DeleteTaskClassItemByID(ctx context.Context, id int) error { + err := dao.db.WithContext(ctx). + Where("id = ?", id). + Delete(&model.TaskClassItem{}).Error + return err +} + +func (dao *TaskClassDAO) DeleteTaskClassByID(ctx context.Context, id int, userID int) error { + // 1. 删除时显式把 user_id 挂到 Model 上，供 GORM 缓存失效插件读取。 + // 2. 业务层已经完成归属校验，这里仍带上 user_id 条件，避免极端并发下误删其它用户数据。 + // 3. 若仍存在 task_items 外键依赖，GORM 会返回数据库错误并回滚，本函数不吞掉该错误。 + res := dao.db.WithContext(ctx). + Model(&model.TaskClass{UserID: &userID}). + Where("id = ? AND user_id = ?", id, userID). + Delete(&model.TaskClass{}) + if res.Error != nil { + return res.Error + } + if res.RowsAffected == 0 { + return respond.WrongTaskClassID + } + return nil +} + +func (dao *TaskClassDAO) BatchUpdateTaskClassItemEmbeddedTime(ctx context.Context, itemIDs []int, updates []*model.TargetTime) error { + if len(itemIDs) == 0 { + return nil + } + if len(itemIDs) != len(updates) { + return errors.New("itemIDs length mismatch updates length") + } + + // 单条 SQL 批量更新：UPDATE ... SET embedded_time = CASE id WHEN ? THEN ? ... END WHERE id IN (?) + caseSQL := "CASE id" + args := make([]any, 0, len(itemIDs)*2) + for i, id := range itemIDs { + caseSQL += " WHEN ? THEN ?" + args = append(args, id, updates[i]) + } + caseSQL += " END" + + res := dao.db.WithContext(ctx). + Model(&model.TaskClassItem{}). + Where("id IN ?", itemIDs). + Update("embedded_time", gorm.Expr(caseSQL, args...)) + + return res.Error +} + +func (dao *TaskClassDAO) ValidateTaskItemIDsBelongToTaskClass(ctx context.Context, taskClassID int, itemIDs []int) (bool, error) { + if len(itemIDs) == 0 { + return true, nil + } + + var count int64 + err := dao.db.WithContext(ctx). + Model(&model.TaskClassItem{}). + Where("id IN ? AND category_id = ?", itemIDs, taskClassID). + Count(&count).Error + if err != nil { + return false, err + } + return count == int64(len(itemIDs)), nil +} + +func (dao *TaskClassDAO) GetTaskClassItemsByIDs(ctx context.Context, itemIDs []int) ([]model.TaskClassItem, error) { + var items []model.TaskClassItem + err := dao.db.WithContext(ctx). + Where("id IN ?", itemIDs). + Find(&items).Error + if err != nil { + return nil, err + } + return items, nil +} diff --git a/backend/services/task_class/rpc/errors.go b/backend/services/task_class/rpc/errors.go new file mode 100644 index 0000000..bb091c1 --- /dev/null +++ b/backend/services/task_class/rpc/errors.go @@ -0,0 +1,83 @@ +package rpc + +import ( + "errors" + "log" + "strings" + + "github.com/LoveLosita/smartflow/backend/respond" + "google.golang.org/genproto/googleapis/rpc/errdetails" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const taskClassErrorDomain = "smartflow.taskclass" + +// grpcErrorFromServiceError 负责把 task-class 内部错误转换为 gRPC status。 +// +// 职责边界： +// 1. respond.Response 保留项目内部 status/info，供 gateway 反解； +// 2. 未分类错误只暴露通用内部错误，详细信息留在服务日志； +// 3. 不在 RPC 层重判业务规则，业务语义仍由 sv/dao 决定。 +func grpcErrorFromServiceError(err error) error { + if err == nil { + return nil + } + var resp respond.Response + if errors.As(err, &resp) { + return grpcErrorFromResponse(resp) + } + log.Printf("task-class rpc internal error: %v", err) + return status.Error(codes.Internal, "task-class service internal error") +} + +func grpcErrorFromResponse(resp respond.Response) error { + code := grpcCodeFromRespondStatus(resp.Status) + message := strings.TrimSpace(resp.Info) + if message == "" { + message = strings.TrimSpace(resp.Status) + } + st := status.New(code, message) + detail := &errdetails.ErrorInfo{ + Domain: taskClassErrorDomain, + Reason: resp.Status, + Metadata: map[string]string{ + "info": resp.Info, + }, + } + withDetails, err := st.WithDetails(detail) + if err != nil { + return st.Err() + } + return withDetails.Err() +} + +func grpcCodeFromRespondStatus(statusValue string) codes.Code { + switch strings.TrimSpace(statusValue) { + case respond.MissingToken.Status, respond.InvalidToken.Status, respond.InvalidClaims.Status, + respond.ErrUnauthorized.Status, respond.WrongTokenType.Status, respond.UserLoggedOut.Status: + return codes.Unauthenticated + case respond.UserTaskClassForbidden.Status, respond.TaskClassNotBelongToUser.Status, + respond.TaskClassItemNotBelongToUser.Status, respond.CourseNotBelongToUser.Status: + return codes.PermissionDenied + case respond.UserTaskClassNotFound.Status, respond.TaskClassItemNotFound.Status: + return codes.NotFound + case respond.MissingParam.Status, respond.WrongParamType.Status, respond.ParamTooLong.Status, + respond.WrongUserID.Status, respond.WrongTaskClassID.Status, respond.WrongTaskID.Status, + respond.InvalidSectionNumber.Status, respond.InvalidWeekOrDayOfWeek.Status, + respond.InvalidSectionRange.Status, respond.MissingParamForAutoScheduling.Status, + respond.InvalidDateRange.Status, respond.TaskClassModeNotAuto.Status, + respond.TimeNotEnoughForAutoScheduling.Status, respond.TaskClassItemNotBelongToTaskClass.Status, + respond.TaskClassItemTryingToInsertOutOfTimeRange.Status, respond.TaskClassItemAlreadyArranged.Status, + respond.CourseAlreadyEmbeddedByOtherTaskBlock.Status, respond.ScheduleConflict.Status, + respond.WrongCourseID.Status, respond.CourseTimeNotMatch.Status, respond.InsertCourseTwice.Status, + respond.WeekOutOfRange.Status, respond.WrongScheduleEventID.Status, + respond.TargetScheduleNotHaveEmbeddedTask.Status, respond.TargetTaskNotEmbeddedInAnySchedule.Status, + respond.TimeOutOfRangeOfThisSemester.Status: + return codes.InvalidArgument + } + if strings.HasPrefix(strings.TrimSpace(statusValue), "5") { + return codes.Internal + } + return codes.InvalidArgument +} diff --git a/backend/services/task_class/rpc/handler.go b/backend/services/task_class/rpc/handler.go new file mode 100644 index 0000000..a575cd4 --- /dev/null +++ b/backend/services/task_class/rpc/handler.go @@ -0,0 +1,221 @@ +package rpc + +import ( + "context" + "encoding/json" + "errors" + + "github.com/LoveLosita/smartflow/backend/model" + "github.com/LoveLosita/smartflow/backend/respond" + "github.com/LoveLosita/smartflow/backend/services/task_class/rpc/pb" + taskclasssv "github.com/LoveLosita/smartflow/backend/services/task_class/sv" + taskclasscontracts "github.com/LoveLosita/smartflow/backend/shared/contracts/taskclass" +) + +const ( + taskClassCreate = 0 + taskClassUpdate = 1 +) + +type Handler struct { + pb.UnimplementedTaskClassServer + svc *taskclasssv.TaskClassService +} + +func NewHandler(svc *taskclasssv.TaskClassService) *Handler { + return &Handler{svc: svc} +} + +// Ping 供调用方在启动期确认 task-class zrpc 已可用。 +func (h *Handler) Ping(ctx context.Context, req *pb.StatusResponse) (*pb.StatusResponse, error) { + if err := h.ensureReady(req); err != nil { + return nil, err + } + return &pb.StatusResponse{}, nil +} + +func (h *Handler) AddTaskClass(ctx context.Context, req *pb.JSONRequest) (*pb.JSONResponse, error) { + if err := h.ensureReady(req); err != nil { + return nil, err + } + var contractReq taskclasscontracts.UpsertTaskClassRequest + if err := json.Unmarshal(req.PayloadJson, &contractReq); err != nil { + return nil, grpcErrorFromServiceError(respond.WrongParamType) + } + err := h.svc.AddOrUpdateTaskClass(ctx, toModelTaskClassRequest(contractReq), contractReq.UserID, taskClassCreate, 0) + return jsonResponse(nil, err) +} + +func (h *Handler) ListTaskClasses(ctx context.Context, req *pb.JSONRequest) (*pb.JSONResponse, error) { + if err := h.ensureReady(req); err != nil { + return nil, err + } + var contractReq taskclasscontracts.UserRequest + if err := json.Unmarshal(req.PayloadJson, &contractReq); err != nil { + return nil, grpcErrorFromServiceError(respond.WrongParamType) + } + data, err := h.svc.GetUserTaskClassInfos(ctx, contractReq.UserID) + return jsonResponse(data, err) +} + +func (h *Handler) GetTaskClass(ctx context.Context, req *pb.JSONRequest) (*pb.JSONResponse, error) { + if err := h.ensureReady(req); err != nil { + return nil, err + } + var contractReq taskclasscontracts.GetTaskClassRequest + if err := json.Unmarshal(req.PayloadJson, &contractReq); err != nil { + return nil, grpcErrorFromServiceError(respond.WrongParamType) + } + data, err := h.svc.GetUserCompleteTaskClass(ctx, contractReq.UserID, contractReq.TaskClassID) + return jsonResponse(data, err) +} + +func (h *Handler) UpdateTaskClass(ctx context.Context, req *pb.JSONRequest) (*pb.JSONResponse, error) { + if err := h.ensureReady(req); err != nil { + return nil, err + } + var contractReq taskclasscontracts.UpsertTaskClassRequest + if err := json.Unmarshal(req.PayloadJson, &contractReq); err != nil { + return nil, grpcErrorFromServiceError(respond.WrongParamType) + } + err := h.svc.AddOrUpdateTaskClass(ctx, toModelTaskClassRequest(contractReq), contractReq.UserID, taskClassUpdate, contractReq.TaskClassID) + return jsonResponse(nil, err) +} + +func (h *Handler) InsertTaskClassItemIntoSchedule(ctx context.Context, req *pb.JSONRequest) (*pb.JSONResponse, error) { + if err := h.ensureReady(req); err != nil { + return nil, err + } + var contractReq taskclasscontracts.InsertTaskClassItemIntoScheduleRequest + if err := json.Unmarshal(req.PayloadJson, &contractReq); err != nil { + return nil, grpcErrorFromServiceError(respond.WrongParamType) + } + err := h.svc.AddTaskClassItemIntoSchedule(ctx, toModelInsertTaskClassItemRequest(contractReq), contractReq.UserID, contractReq.TaskItemID) + return jsonResponse(nil, err) +} + +func (h *Handler) DeleteTaskClassItem(ctx context.Context, req *pb.JSONRequest) (*pb.JSONResponse, error) { + if err := h.ensureReady(req); err != nil { + return nil, err + } + var contractReq taskclasscontracts.DeleteTaskClassItemRequest + if err := json.Unmarshal(req.PayloadJson, &contractReq); err != nil { + return nil, grpcErrorFromServiceError(respond.WrongParamType) + } + err := h.svc.DeleteTaskClassItem(ctx, contractReq.UserID, contractReq.TaskItemID) + return jsonResponse(nil, err) +} + +func (h *Handler) DeleteTaskClass(ctx context.Context, req *pb.JSONRequest) (*pb.JSONResponse, error) { + if err := h.ensureReady(req); err != nil { + return nil, err + } + var contractReq taskclasscontracts.DeleteTaskClassRequest + if err := json.Unmarshal(req.PayloadJson, &contractReq); err != nil { + return nil, grpcErrorFromServiceError(respond.WrongParamType) + } + err := h.svc.DeleteTaskClass(ctx, contractReq.UserID, contractReq.TaskClassID) + return jsonResponse(nil, err) +} + +func (h *Handler) ApplyBatchIntoSchedule(ctx context.Context, req *pb.JSONRequest) (*pb.JSONResponse, error) { + if err := h.ensureReady(req); err != nil { + return nil, err + } + var contractReq taskclasscontracts.ApplyBatchIntoScheduleRequest + if err := json.Unmarshal(req.PayloadJson, &contractReq); err != nil { + return nil, grpcErrorFromServiceError(respond.WrongParamType) + } + err := h.svc.BatchApplyPlans(ctx, contractReq.TaskClassID, contractReq.UserID, toModelBatchRequest(contractReq)) + return jsonResponse(nil, err) +} + +func (h *Handler) ensureReady(req any) error { + if h == nil || h.svc == nil { + return grpcErrorFromServiceError(errors.New("task-class service dependency not initialized")) + } + if req == nil { + return grpcErrorFromServiceError(respond.MissingParam) + } + return nil +} + +func jsonResponse(value any, err error) (*pb.JSONResponse, error) { + if err != nil { + return nil, grpcErrorFromServiceError(err) + } + raw, err := json.Marshal(value) + if err != nil { + return nil, grpcErrorFromServiceError(err) + } + return &pb.JSONResponse{DataJson: raw}, nil +} + +func toModelTaskClassRequest(req taskclasscontracts.UpsertTaskClassRequest) *model.UserAddTaskClassRequest { + items := make([]model.UserAddTaskClassItemRequest, 0, len(req.Items)) + for _, item := range req.Items { + items = append(items, model.UserAddTaskClassItemRequest{ + ID: item.ID, + Order: item.Order, + Content: item.Content, + EmbeddedTime: toModelTargetTime(item.EmbeddedTime), + }) + } + return &model.UserAddTaskClassRequest{ + Name: req.Name, + StartDate: req.StartDate, + EndDate: req.EndDate, + Mode: req.Mode, + SubjectType: req.SubjectType, + DifficultyLevel: req.DifficultyLevel, + CognitiveIntensity: req.CognitiveIntensity, + Config: model.UserAddTaskClassConfig{ + TotalSlots: req.Config.TotalSlots, + AllowFillerCourse: req.Config.AllowFillerCourse, + Strategy: req.Config.Strategy, + ExcludedSlots: append([]int(nil), req.Config.ExcludedSlots...), + ExcludedDaysOfWeek: append([]int(nil), req.Config.ExcludedDaysOfWeek...), + }, + Items: items, + } +} + +func toModelTargetTime(value *taskclasscontracts.TargetTime) *model.TargetTime { + if value == nil { + return nil + } + return &model.TargetTime{ + Week: value.Week, + DayOfWeek: value.DayOfWeek, + SectionFrom: value.SectionFrom, + SectionTo: value.SectionTo, + } +} + +func toModelInsertTaskClassItemRequest(req taskclasscontracts.InsertTaskClassItemIntoScheduleRequest) *model.UserInsertTaskClassItemToScheduleRequest { + return &model.UserInsertTaskClassItemToScheduleRequest{ + Week: req.Week, + DayOfWeek: req.DayOfWeek, + StartSection: req.StartSection, + EndSection: req.EndSection, + EmbedCourseEventID: req.EmbedCourseEventID, + } +} + +func toModelBatchRequest(req taskclasscontracts.ApplyBatchIntoScheduleRequest) *model.UserInsertTaskClassItemToScheduleRequestBatch { + items := make([]model.SingleTaskClassItem, 0, len(req.Items)) + for _, item := range req.Items { + items = append(items, model.SingleTaskClassItem{ + TaskItemID: item.TaskItemID, + Week: item.Week, + DayOfWeek: item.DayOfWeek, + StartSection: item.StartSection, + EndSection: item.EndSection, + EmbedCourseEventID: item.EmbedCourseEventID, + }) + } + return &model.UserInsertTaskClassItemToScheduleRequestBatch{ + TaskClassID: req.TaskClassID, + Items: items, + } +} diff --git a/backend/services/task_class/rpc/pb/task_class.pb.go b/backend/services/task_class/rpc/pb/task_class.pb.go new file mode 100644 index 0000000..b6fe02e --- /dev/null +++ b/backend/services/task_class/rpc/pb/task_class.pb.go @@ -0,0 +1,39 @@ +package pb + +import proto "github.com/golang/protobuf/proto" + +var _ = proto.Marshal + +const _ = proto.ProtoPackageIsVersion3 + +type JSONRequest struct { + PayloadJson []byte `protobuf:"bytes,1,opt,name=payload_json,json=payloadJson,proto3" json:"payload_json,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *JSONRequest) Reset() { *m = JSONRequest{} } +func (m *JSONRequest) String() string { return proto.CompactTextString(m) } +func (*JSONRequest) ProtoMessage() {} + +type JSONResponse struct { + DataJson []byte `protobuf:"bytes,1,opt,name=data_json,json=dataJson,proto3" json:"data_json,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *JSONResponse) Reset() { *m = JSONResponse{} } +func (m *JSONResponse) String() string { return proto.CompactTextString(m) } +func (*JSONResponse) ProtoMessage() {} + +type StatusResponse struct { + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *StatusResponse) Reset() { *m = StatusResponse{} } +func (m *StatusResponse) String() string { return proto.CompactTextString(m) } +func (*StatusResponse) ProtoMessage() {} diff --git a/backend/services/task_class/rpc/pb/task_class_grpc.pb.go b/backend/services/task_class/rpc/pb/task_class_grpc.pb.go new file mode 100644 index 0000000..12ca188 --- /dev/null +++ b/backend/services/task_class/rpc/pb/task_class_grpc.pb.go @@ -0,0 +1,191 @@ +package pb + +import ( + context "context" + + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +const ( + TaskClass_Ping_FullMethodName = "/smartflow.taskclass.TaskClass/Ping" + TaskClass_AddTaskClass_FullMethodName = "/smartflow.taskclass.TaskClass/AddTaskClass" + TaskClass_ListTaskClasses_FullMethodName = "/smartflow.taskclass.TaskClass/ListTaskClasses" + TaskClass_GetTaskClass_FullMethodName = "/smartflow.taskclass.TaskClass/GetTaskClass" + TaskClass_UpdateTaskClass_FullMethodName = "/smartflow.taskclass.TaskClass/UpdateTaskClass" + TaskClass_InsertTaskClassItemIntoSchedule_FullMethodName = "/smartflow.taskclass.TaskClass/InsertTaskClassItemIntoSchedule" + TaskClass_DeleteTaskClassItem_FullMethodName = "/smartflow.taskclass.TaskClass/DeleteTaskClassItem" + TaskClass_DeleteTaskClass_FullMethodName = "/smartflow.taskclass.TaskClass/DeleteTaskClass" + TaskClass_ApplyBatchIntoSchedule_FullMethodName = "/smartflow.taskclass.TaskClass/ApplyBatchIntoSchedule" +) + +type TaskClassClient interface { + Ping(ctx context.Context, in *StatusResponse, opts ...grpc.CallOption) (*StatusResponse, error) + AddTaskClass(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) + ListTaskClasses(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) + GetTaskClass(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) + UpdateTaskClass(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) + InsertTaskClassItemIntoSchedule(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) + DeleteTaskClassItem(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) + DeleteTaskClass(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) + ApplyBatchIntoSchedule(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) +} + +type taskClassClient struct { + cc grpc.ClientConnInterface +} + +func NewTaskClassClient(cc grpc.ClientConnInterface) TaskClassClient { + return &taskClassClient{cc} +} + +func (c *taskClassClient) Ping(ctx context.Context, in *StatusResponse, opts ...grpc.CallOption) (*StatusResponse, error) { + out := new(StatusResponse) + err := c.cc.Invoke(ctx, TaskClass_Ping_FullMethodName, in, out, opts...) + return out, err +} + +func (c *taskClassClient) AddTaskClass(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) { + out := new(JSONResponse) + err := c.cc.Invoke(ctx, TaskClass_AddTaskClass_FullMethodName, in, out, opts...) + return out, err +} + +func (c *taskClassClient) ListTaskClasses(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) { + out := new(JSONResponse) + err := c.cc.Invoke(ctx, TaskClass_ListTaskClasses_FullMethodName, in, out, opts...) + return out, err +} + +func (c *taskClassClient) GetTaskClass(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) { + out := new(JSONResponse) + err := c.cc.Invoke(ctx, TaskClass_GetTaskClass_FullMethodName, in, out, opts...) + return out, err +} + +func (c *taskClassClient) UpdateTaskClass(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) { + out := new(JSONResponse) + err := c.cc.Invoke(ctx, TaskClass_UpdateTaskClass_FullMethodName, in, out, opts...) + return out, err +} + +func (c *taskClassClient) InsertTaskClassItemIntoSchedule(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) { + out := new(JSONResponse) + err := c.cc.Invoke(ctx, TaskClass_InsertTaskClassItemIntoSchedule_FullMethodName, in, out, opts...) + return out, err +} + +func (c *taskClassClient) DeleteTaskClassItem(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) { + out := new(JSONResponse) + err := c.cc.Invoke(ctx, TaskClass_DeleteTaskClassItem_FullMethodName, in, out, opts...) + return out, err +} + +func (c *taskClassClient) DeleteTaskClass(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) { + out := new(JSONResponse) + err := c.cc.Invoke(ctx, TaskClass_DeleteTaskClass_FullMethodName, in, out, opts...) + return out, err +} + +func (c *taskClassClient) ApplyBatchIntoSchedule(ctx context.Context, in *JSONRequest, opts ...grpc.CallOption) (*JSONResponse, error) { + out := new(JSONResponse) + err := c.cc.Invoke(ctx, TaskClass_ApplyBatchIntoSchedule_FullMethodName, in, out, opts...) + return out, err +} + +type TaskClassServer interface { + Ping(context.Context, *StatusResponse) (*StatusResponse, error) + AddTaskClass(context.Context, *JSONRequest) (*JSONResponse, error) + ListTaskClasses(context.Context, *JSONRequest) (*JSONResponse, error) + GetTaskClass(context.Context, *JSONRequest) (*JSONResponse, error) + UpdateTaskClass(context.Context, *JSONRequest) (*JSONResponse, error) + InsertTaskClassItemIntoSchedule(context.Context, *JSONRequest) (*JSONResponse, error) + DeleteTaskClassItem(context.Context, *JSONRequest) (*JSONResponse, error) + DeleteTaskClass(context.Context, *JSONRequest) (*JSONResponse, error) + ApplyBatchIntoSchedule(context.Context, *JSONRequest) (*JSONResponse, error) +} + +type UnimplementedTaskClassServer struct{} + +func (UnimplementedTaskClassServer) Ping(context.Context, *StatusResponse) (*StatusResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented") +} +func (UnimplementedTaskClassServer) AddTaskClass(context.Context, *JSONRequest) (*JSONResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method AddTaskClass not implemented") +} +func (UnimplementedTaskClassServer) ListTaskClasses(context.Context, *JSONRequest) (*JSONResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method ListTaskClasses not implemented") +} +func (UnimplementedTaskClassServer) GetTaskClass(context.Context, *JSONRequest) (*JSONResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method GetTaskClass not implemented") +} +func (UnimplementedTaskClassServer) UpdateTaskClass(context.Context, *JSONRequest) (*JSONResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method UpdateTaskClass not implemented") +} +func (UnimplementedTaskClassServer) InsertTaskClassItemIntoSchedule(context.Context, *JSONRequest) (*JSONResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method InsertTaskClassItemIntoSchedule not implemented") +} +func (UnimplementedTaskClassServer) DeleteTaskClassItem(context.Context, *JSONRequest) (*JSONResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method DeleteTaskClassItem not implemented") +} +func (UnimplementedTaskClassServer) DeleteTaskClass(context.Context, *JSONRequest) (*JSONResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method DeleteTaskClass not implemented") +} +func (UnimplementedTaskClassServer) ApplyBatchIntoSchedule(context.Context, *JSONRequest) (*JSONResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method ApplyBatchIntoSchedule not implemented") +} + +func RegisterTaskClassServer(s grpc.ServiceRegistrar, srv TaskClassServer) { + s.RegisterService(&TaskClass_ServiceDesc, srv) +} + +func _TaskClass_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(StatusResponse) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(TaskClassServer).Ping(ctx, in) + } + info := &grpc.UnaryServerInfo{Server: srv, FullMethod: TaskClass_Ping_FullMethodName} + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(TaskClassServer).Ping(ctx, req.(*StatusResponse)) + } + return interceptor(ctx, in, info, handler) +} + +func _TaskClass_JSON_Handler(fullMethod string, invoke func(TaskClassServer, context.Context, *JSONRequest) (*JSONResponse, error)) grpc.MethodHandler { + return func(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(JSONRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return invoke(srv.(TaskClassServer), ctx, in) + } + info := &grpc.UnaryServerInfo{Server: srv, FullMethod: fullMethod} + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return invoke(srv.(TaskClassServer), ctx, req.(*JSONRequest)) + } + return interceptor(ctx, in, info, handler) + } +} + +var TaskClass_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "smartflow.taskclass.TaskClass", + HandlerType: (*TaskClassServer)(nil), + Methods: []grpc.MethodDesc{ + {MethodName: "Ping", Handler: _TaskClass_Ping_Handler}, + {MethodName: "AddTaskClass", Handler: _TaskClass_JSON_Handler(TaskClass_AddTaskClass_FullMethodName, TaskClassServer.AddTaskClass)}, + {MethodName: "ListTaskClasses", Handler: _TaskClass_JSON_Handler(TaskClass_ListTaskClasses_FullMethodName, TaskClassServer.ListTaskClasses)}, + {MethodName: "GetTaskClass", Handler: _TaskClass_JSON_Handler(TaskClass_GetTaskClass_FullMethodName, TaskClassServer.GetTaskClass)}, + {MethodName: "UpdateTaskClass", Handler: _TaskClass_JSON_Handler(TaskClass_UpdateTaskClass_FullMethodName, TaskClassServer.UpdateTaskClass)}, + {MethodName: "InsertTaskClassItemIntoSchedule", Handler: _TaskClass_JSON_Handler(TaskClass_InsertTaskClassItemIntoSchedule_FullMethodName, TaskClassServer.InsertTaskClassItemIntoSchedule)}, + {MethodName: "DeleteTaskClassItem", Handler: _TaskClass_JSON_Handler(TaskClass_DeleteTaskClassItem_FullMethodName, TaskClassServer.DeleteTaskClassItem)}, + {MethodName: "DeleteTaskClass", Handler: _TaskClass_JSON_Handler(TaskClass_DeleteTaskClass_FullMethodName, TaskClassServer.DeleteTaskClass)}, + {MethodName: "ApplyBatchIntoSchedule", Handler: _TaskClass_JSON_Handler(TaskClass_ApplyBatchIntoSchedule_FullMethodName, TaskClassServer.ApplyBatchIntoSchedule)}, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "task_class.proto", +} diff --git a/backend/services/task_class/rpc/server.go b/backend/services/task_class/rpc/server.go new file mode 100644 index 0000000..23f7417 --- /dev/null +++ b/backend/services/task_class/rpc/server.go @@ -0,0 +1,60 @@ +package rpc + +import ( + "errors" + "strings" + "time" + + "github.com/LoveLosita/smartflow/backend/services/task_class/rpc/pb" + taskclasssv "github.com/LoveLosita/smartflow/backend/services/task_class/sv" + "github.com/zeromicro/go-zero/core/service" + "github.com/zeromicro/go-zero/zrpc" + "google.golang.org/grpc" +) + +const ( + defaultListenOn = "0.0.0.0:9086" + defaultTimeout = 6 * time.Second +) + +type ServerOptions struct { + ListenOn string + Timeout time.Duration + Service *taskclasssv.TaskClassService +} + +// NewServer 创建 task-class zrpc 服务端。 +// +// 职责边界： +// 1. 只负责 zrpc server 配置与 gRPC handler 注册； +// 2. 不创建数据库、Redis 或业务服务，它们由 cmd/task-class 管理； +// 3. 返回 listenOn 供进程入口打印启动日志。 +func NewServer(opts ServerOptions) (*zrpc.RpcServer, string, error) { + if opts.Service == nil { + return nil, "", errors.New("task-class service dependency not initialized") + } + + listenOn := strings.TrimSpace(opts.ListenOn) + if listenOn == "" { + listenOn = defaultListenOn + } + timeout := opts.Timeout + if timeout <= 0 { + timeout = defaultTimeout + } + + server, err := zrpc.NewServer(zrpc.RpcServerConf{ + ServiceConf: service.ServiceConf{ + Name: "task-class.rpc", + Mode: service.DevMode, + }, + ListenOn: listenOn, + Timeout: int64(timeout / time.Millisecond), + }, func(grpcServer *grpc.Server) { + pb.RegisterTaskClassServer(grpcServer, NewHandler(opts.Service)) + }) + if err != nil { + return nil, "", err + } + return server, listenOn, nil +} diff --git a/backend/services/task_class/rpc/task_class.proto b/backend/services/task_class/rpc/task_class.proto new file mode 100644 index 0000000..5a95242 --- /dev/null +++ b/backend/services/task_class/rpc/task_class.proto @@ -0,0 +1,28 @@ +syntax = "proto3"; + +package smartflow.taskclass; + +option go_package = "github.com/LoveLosita/smartflow/backend/services/task_class/rpc/pb"; + +service TaskClass { + rpc Ping(StatusResponse) returns (StatusResponse); + rpc AddTaskClass(JSONRequest) returns (JSONResponse); + rpc ListTaskClasses(JSONRequest) returns (JSONResponse); + rpc GetTaskClass(JSONRequest) returns (JSONResponse); + rpc UpdateTaskClass(JSONRequest) returns (JSONResponse); + rpc InsertTaskClassItemIntoSchedule(JSONRequest) returns (JSONResponse); + rpc DeleteTaskClassItem(JSONRequest) returns (JSONResponse); + rpc DeleteTaskClass(JSONRequest) returns (JSONResponse); + rpc ApplyBatchIntoSchedule(JSONRequest) returns (JSONResponse); +} + +message JSONRequest { + bytes payload_json = 1; +} + +message JSONResponse { + bytes data_json = 1; +} + +message StatusResponse { +} diff --git a/backend/services/task_class/sv/service.go b/backend/services/task_class/sv/service.go new file mode 100644 index 0000000..71c6a04 --- /dev/null +++ b/backend/services/task_class/sv/service.go @@ -0,0 +1,553 @@ +package sv + +import ( + "context" + "errors" + "fmt" + "log" + "sort" + "time" + + "github.com/LoveLosita/smartflow/backend/conv" + rootdao "github.com/LoveLosita/smartflow/backend/dao" + "github.com/LoveLosita/smartflow/backend/model" + "github.com/LoveLosita/smartflow/backend/respond" + taskclassdao "github.com/LoveLosita/smartflow/backend/services/task_class/dao" + "github.com/go-redis/redis/v8" + "gorm.io/gorm" +) + +type TaskClassService struct { + // 这里可以添加数据库连接或其他依赖 + taskClassRepo *taskclassdao.TaskClassDAO + cacheRepo *rootdao.CacheDAO + scheduleRepo *rootdao.ScheduleDAO + repoManager *rootdao.RepoManager // 统一管理多个 DAO 的事务 +} + +func NewTaskClassService(taskClassRepo *taskclassdao.TaskClassDAO, cacheRepo *rootdao.CacheDAO, scheduleRepo *rootdao.ScheduleDAO, manager *rootdao.RepoManager) *TaskClassService { + return &TaskClassService{ + taskClassRepo: taskClassRepo, + cacheRepo: cacheRepo, + scheduleRepo: scheduleRepo, + repoManager: manager, + } +} + +// AddOrUpdateTaskClass 为指定用户添加任务类 +func (sv *TaskClassService) AddOrUpdateTaskClass(ctx context.Context, req *model.UserAddTaskClassRequest, userID int, method int, targetTaskClassID int) error { + //1.先校验参数 + if req.Mode == "auto" { + if req.StartDate == "" || req.EndDate == "" { + return respond.MissingParamForAutoScheduling + } + st, err := time.Parse("2006-01-02", req.StartDate) + if err != nil { + return respond.WrongParamType + } + ed, err := time.Parse("2006-01-02", req.EndDate) + if err != nil { + return respond.WrongParamType + } + if st.After(ed) { + return respond.InvalidDateRange + } + } + if req.Mode == "" || req.Name == "" || len(req.Items) == 0 { + return respond.MissingParam + } + // 1. excluded_slots 属于“半天块索引”，每个索引映射 2 节（1->1-2，...，6->11-12）； + // 2. 若允许 7~12，会在粗排网格展开时产生越界节次，触发运行时 panic； + // 3. 这里统一在写入入口拦截，避免脏数据落库后污染后续排程链路。 + for _, slot := range req.Config.ExcludedSlots { + if slot < 1 || slot > 6 { + return respond.WrongParamType + } + } + // 1. excluded_days_of_week 表示“整天不可排”的硬约束，粗排时会直接整天屏蔽； + // 2. 只允许 1~7，对应周一到周日； + // 3. 若写入非法值，会导致粗排过滤口径和前端展示口径不一致，因此入口直接拦截。 + for _, dayOfWeek := range req.Config.ExcludedDaysOfWeek { + if dayOfWeek < 1 || dayOfWeek > 7 { + return respond.WrongParamType + } + } + //2.写数据库（事务内） + if err := sv.taskClassRepo.Transaction(func(txDAO *taskclassdao.TaskClassDAO) error { + taskClass, items, err := conv.ProcessUserAddTaskClassRequest(req, userID) + if err != nil { + return err + } + if method == 1 { // 更新操作 + taskClass.ID = targetTaskClassID + } + + taskClassID, err := txDAO.AddOrUpdateTaskClass(userID, taskClass) + if err != nil { + return err + } + + for i := range items { + items[i].CategoryID = &taskClassID + } + if err := txDAO.AddOrUpdateTaskClassItems(userID, items); err != nil { + return err + } + return nil + }); err != nil { + return err + } + + return nil +} + +func (sv *TaskClassService) GetUserTaskClassInfos(ctx context.Context, userID int) (*model.UserGetTaskClassesResponse, error) { + //1.先查询redis + list, err := sv.cacheRepo.GetTaskClassList(ctx, userID) + if err == nil { + //命中缓存 + return list, nil + } else if !errors.Is(err, redis.Nil) { //不是缓存未命中错误，说明redis可能炸了，照常放行 + log.Println("redis获取任务分类列表失败:", err) + } + //2.缓存未命中，查询数据库 + taskClasses, err := sv.taskClassRepo.GetUserTaskClasses(userID) + if err != nil { + return nil, err + } + resp := conv.TaskClassModelToResponse(taskClasses) + //3.写入缓存 + err = sv.cacheRepo.AddTaskClassList(ctx, userID, resp) + if err != nil { + return nil, err + } + return resp, nil +} + +func (sv *TaskClassService) GetUserCompleteTaskClass(ctx context.Context, userID int, taskClassID int) (*model.UserAddTaskClassRequest, error) { + //1.查询数据库 + taskClass, err := sv.taskClassRepo.GetCompleteTaskClassByID(ctx, taskClassID, userID) + if err != nil { + return nil, err + } + //2.转换为响应结构体 + resp, err := conv.ProcessUserGetCompleteTaskClassRequest(taskClass) + if err != nil { + return nil, err + } + return resp, nil +} + +func (sv *TaskClassService) AddTaskClassItemIntoSchedule(ctx context.Context, req *model.UserInsertTaskClassItemToScheduleRequest, userID int, taskID int) error { + //1.先验证任务块归属 + taskClassID, err := sv.taskClassRepo.GetTaskClassIDByTaskItemID(ctx, taskID) //通过任务块ID获取所属任务类ID + if err != nil { + return err + } + ownerID, err := sv.taskClassRepo.GetTaskClassUserIDByID(ctx, taskClassID) //通过任务类ID获取所属用户ID + if err != nil { + return err + } + if ownerID != userID { + return respond.TaskClassItemNotBelongToUser + } + //2.再检查任务块本身是否已经被安排 + result, err := sv.taskClassRepo.IfTaskClassItemArranged(ctx, taskID) + if err != nil { + return err + } + if result { + return respond.TaskClassItemAlreadyArranged + } + //3.取出任务块信息 + taskItem, err := sv.taskClassRepo.GetTaskClassItemByID(ctx, taskID) //通过任务块ID获取任务块信息 + if err != nil { + return err + } + //更新TaskClassItem的embedded_time字段 + taskItem.EmbeddedTime = &model.TargetTime{ + DayOfWeek: req.DayOfWeek, + Week: req.Week, + SectionFrom: req.StartSection, + SectionTo: req.EndSection, + } + //3.判断是否嵌入课程 + if req.EmbedCourseEventID != 0 { + //先检查看课程是否存在、是否归属该用户以及是否已经被嵌入了其他任务块 + courseOwnerID, err := sv.scheduleRepo.GetCourseUserIDByID(ctx, req.EmbedCourseEventID) + if err != nil { + return err + } + if courseOwnerID != userID { + return respond.CourseNotBelongToUser + } + //再检查用户给的时间是否和课程的时间匹配(目前逻辑是给的区间必须完全匹配) + match, err := sv.scheduleRepo.IsCourseTimeMatch(ctx, req.EmbedCourseEventID, req.Week, req.DayOfWeek, req.StartSection, req.EndSection) + if err != nil { + return err + } + if !match { + return respond.CourseTimeNotMatch + } + //查询对应时段的课程是否已被其他任务块嵌入了（目前业务限制：一个课程只能被一个任务块嵌入，但是目前设计是支持多个任务块嵌入一节课的，只要放得下） + isEmbedded, err := sv.scheduleRepo.IsCourseEmbeddedByOtherTaskBlock(ctx, req.EmbedCourseEventID, req.StartSection, req.EndSection) + if err != nil { + return err + } + if isEmbedded { + return respond.CourseAlreadyEmbeddedByOtherTaskBlock + } + //嵌入课程，直接更新日程表对应时段的 embedded_task_id 字段 + err = sv.scheduleRepo.EmbedTaskIntoSchedule(req.StartSection, req.EndSection, req.DayOfWeek, req.Week, userID, taskID) + if err != nil { + return err + } + //更新任务块的 embedded_time 字段 + err = sv.taskClassRepo.UpdateTaskClassItemEmbeddedTime(ctx, taskID, taskItem.EmbeddedTime) + if err != nil { + return err + } + return nil + } + //4.否则构造Schedule模型 + sections := make([]int, 0, req.EndSection-req.StartSection+1) + schedules, scheduleEvent, err := conv.UserInsertTaskItemRequestToModel(req, taskItem, nil, userID, req.StartSection, req.EndSection) + if err != nil { + return err + } + //将节次区间转换为节次切片，方便后续检查冲突 + for section := req.StartSection; section <= req.EndSection; section++ { + sections = append(sections, section) + } + //4.1 统一检查冲突（避免逐条查库） + conflict, err := sv.scheduleRepo.HasUserScheduleConflict(ctx, userID, req.Week, req.DayOfWeek, sections) + if err != nil { + return err + } + if conflict { + return respond.ScheduleConflict + } + // 5. 写入数据库（通过 RepoManager 统一管理事务） + // 这里的 sv.daoManager 是你在初始化 Service 时注入的全局 RepoManager 实例 + if err := sv.repoManager.Transaction(ctx, func(txM *rootdao.RepoManager) error { + // 5.1 使用事务中的 ScheduleRepo 插入 Event + // 💡 这里的 txM.Schedule 已经注入了事务句柄 + //此处要将req中的起始section以及第几周、星期几转换成绝对时间，存入scheduleEvent的StartTime和EndTime字段中，方便后续查询和冲突检查 + st, ed, err := conv.RelativeTimeToRealTime(req.Week, req.DayOfWeek, req.StartSection, req.EndSection) + if err != nil { + return err + } + scheduleEvent.StartTime = st + scheduleEvent.EndTime = ed + eventID, err := txM.Schedule.AddScheduleEvent(scheduleEvent) + if err != nil { + return err // 触发回滚 + } + // 5.2 关联 ID（纯内存操作，无需 tx） + for i := range schedules { + schedules[i].EventID = eventID + } + // 5.3 使用事务中的 ScheduleRepo 批量插入原子槽位 + // 💡 如果这里因为外键或唯一索引报错，5.1 的 Event 也会被撤回 + if _, err = txM.Schedule.AddSchedules(schedules); err != nil { + return err // 触发回滚 + } + // 5.4 使用事务中的 TaskRepo 更新任务状态 + // 💡 这里的 txM.Task 取代了你原来的 txDAO + if err := txM.TaskClass.UpdateTaskClassItemEmbeddedTime(ctx, taskID, taskItem.EmbeddedTime); err != nil { + return err // 触发回滚 + } + return nil + }); err != nil { + // 这里处理最终的错误返回，比如 respond.Error + return err + } + return nil +} + +func (sv *TaskClassService) DeleteTaskClassItem(ctx context.Context, userID int, taskItemID int) error { + //1.先验证任务块归属 + taskClassID, err := sv.taskClassRepo.GetTaskClassIDByTaskItemID(ctx, taskItemID) //通过任务块ID获取所属任务类ID + if err != nil { + return err + } + ownerID, err := sv.taskClassRepo.GetTaskClassUserIDByID(ctx, taskClassID) //通过任务类ID获取所属用户ID + if err != nil { + return err + } + if ownerID != userID { + return respond.TaskClassItemNotBelongToUser + } + //2.如果该任务块已经被安排了，先解除安排，再删除任务块（事务） + if err := sv.repoManager.Transaction(ctx, func(txM *rootdao.RepoManager) error { + //2.1.先检查该任务块是否已经被安排了 + arranged, err := txM.TaskClass.IfTaskClassItemArranged(ctx, taskItemID) + if err != nil { + return err + } + if arranged { + //2.2.如果已经被安排了，先解除安排 + //先扫schedules找到该task_item_id并删除 + _, txErr := txM.Schedule.FindEmbeddedTaskIDAndDeleteIt(ctx, taskItemID) + //2.3.再将task_items表的embedded_time字段设置为null + txErr = txM.TaskClass.DeleteTaskClassItemEmbeddedTime(ctx, taskItemID) + if txErr != nil { + return txErr + } + //再删除schedule_event表中对应的事件 + txErr = txM.Schedule.DeleteScheduleEventByTaskItemID(ctx, taskItemID) + if txErr != nil { + return txErr + } + } + //2.4.最后删除任务块 + err = txM.TaskClass.DeleteTaskClassItemByID(ctx, taskItemID) + if err != nil { + return err + } + return nil + }); err != nil { + return err + } + return nil +} + +func (sv *TaskClassService) DeleteTaskClass(ctx context.Context, userID int, taskClassID int) error { + //1.先验证任务类归属 + ownerID, err := sv.taskClassRepo.GetTaskClassUserIDByID(ctx, taskClassID) //通过任务类ID获取所属用户ID + if err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return respond.WrongTaskClassID + } + return err + } + if ownerID != userID { + return respond.TaskClassNotBelongToUser + } + //2.删除任务类（事务） + err = sv.taskClassRepo.DeleteTaskClassByID(ctx, taskClassID, userID) + if err != nil { + return err + } + return nil +} + +// GetCompleteTaskClassByID 获取任务类完整详情（含关联的 TaskClassItem 列表）。 +// +// 职责边界： +// 1) 直接委托 DAO 层查询，不做额外业务逻辑； +// 2) 主要供 Agent 排程链路使用，获取 Items 用于 materialize 节点映射。 +func (sv *TaskClassService) GetCompleteTaskClassByID(ctx context.Context, taskClassID, userID int) (*model.TaskClass, error) { + return sv.taskClassRepo.GetCompleteTaskClassByID(ctx, taskClassID, userID) +} + +func (sv *TaskClassService) BatchApplyPlans(ctx context.Context, taskClassID int, userID int, plans *model.UserInsertTaskClassItemToScheduleRequestBatch) error { + //1.通过任务类id获取任务类详情 + taskClass, err := sv.taskClassRepo.GetCompleteTaskClassByID(ctx, taskClassID, userID) + if err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return respond.WrongTaskClassID + } + return err + } + //2.校验任务类的参数是否合法 + if taskClass == nil { + return respond.WrongTaskClassID + } + if *taskClass.Mode != "auto" { + return respond.TaskClassModeNotAuto + } + //3.获取任务类安排的时间范围内的全部周数信息(左右边界不足一周的情况也要算作一周)，用于下方冲突检查 + startWeekTime := conv.CalculateFirstDayOfWeek(*taskClass.StartDate) + endWeekTime := conv.CalculateLastDayOfWeek(*taskClass.EndDate) + schedules, err := sv.scheduleRepo.GetUserSchedulesByTimeRange(ctx, userID, startWeekTime, endWeekTime) + if err != nil { + return err + } + startWeek, _, err := conv.RealDateToRelativeDate(startWeekTime.Format("2006-01-02")) + if err != nil { + return err + } + endWeek, _, err := conv.RealDateToRelativeDate(endWeekTime.Format("2006-01-02")) + if err != nil { + return err + } + //4.统一检查冲突（避免逐条查库） + //先将日程放入一个map中，key是"周-星期-节次"，value是课程信息，方便后续检查冲突 + courseMap := make(map[string]model.Schedule) + for _, schedule := range schedules { + key := fmt.Sprintf("%d-%d-%d", schedule.Week, schedule.DayOfWeek, schedule.Section) + courseMap[key] = schedule + } + //再遍历每个任务块的安排时间，检查是否和课程冲突（目前逻辑是只要有一个时段冲突就算冲突，后续可以优化为统计冲突的时段数量，或者提供具体的冲突时段信息） + for _, plan := range plans.Items { + if plan.Week < startWeek || plan.Week > endWeek { + return respond.TaskClassItemTryingToInsertOutOfTimeRange + } + for section := plan.StartSection; section <= plan.EndSection; section++ { + key := fmt.Sprintf("%d-%d-%d", plan.Week, plan.DayOfWeek, section) + // 如果课程存在，并且满足以下任一条件则认为冲突： + // 1. 课程时段已经被其他任务块嵌入了（不允许多个任务块嵌入同一课程） + // 2. 当前时段的课的EventID与用户计划中指定的EmbedCourseEventID不匹配（说明用户计划要嵌入的课程和当前时段的课不是同一节） + // 3. 用户计划中没有指定EmbedCourseEventID（即EmbedCourseEventID为0），但当前时段有课（不允许在有课的时段安排任务块） + // 4. 当前时段的课不允许被嵌入（即使用户计划中指定了EmbedCourseEventID，但如果课程本身不允许被嵌入了，也算冲突） + if course, exists := courseMap[key]; exists && ((plan.EmbedCourseEventID != 0 && course.EmbeddedTask != nil) || + (plan.EmbedCourseEventID != course.EventID) || plan.EmbedCourseEventID == 0 || !course.Event.CanBeEmbedded) { + return respond.ScheduleConflict + } + } + } + //5.分流批量写入数据库（通过 RepoManager 统一管理事务） + //先分流 + toEmbed := make([]model.SingleTaskClassItem, 0) //需要嵌入课程的任务块 + toNormal := make([]model.SingleTaskClassItem, 0) //需要新建日程的任务块 + for _, item := range plans.Items { + if item.EmbedCourseEventID != 0 { + toEmbed = append(toEmbed, item) + } else { + toNormal = append(toNormal, item) + } + } + //再开事务批量写库 + if err := sv.repoManager.Transaction(ctx, func(txM *rootdao.RepoManager) error { + //5.1 先处理需要嵌入课程的任务块 + //先提取出需要嵌入的课程ID和TaskItemID列表 + courseIDs := make([]int, 0, len(toEmbed)) + for _, item := range toEmbed { + courseIDs = append(courseIDs, item.EmbedCourseEventID) + } + itemIDs := make([]int, 0, len(toEmbed)) + for _, item := range toEmbed { + itemIDs = append(itemIDs, item.TaskItemID) + } + //检查任务块本身是否已经被安排 + result, err := sv.taskClassRepo.BatchCheckIfTaskClassItemsArranged(ctx, itemIDs) + if err != nil { + return err + } + if result { + return respond.TaskClassItemAlreadyArranged + } + //验证一下plans中的taskItemID确实都属于这个用户和这个任务类（避免用户恶意构造请求把别的用户的任务块或者不属于任何任务类的任务块也安排了） + //同时也能检查是否重复 + result, err = sv.taskClassRepo.ValidateTaskItemIDsBelongToTaskClass(ctx, taskClassID, itemIDs) + if err != nil { + return err + } + if !result { + return respond.TaskClassItemNotBelongToTaskClass + } + //批量更新日程表中对应课程的embedded_task_id字段（目前业务限制：一个课程只能被一个任务块嵌入了，所以直接批量更新，不用担心覆盖问题） + err = txM.Schedule.BatchEmbedTaskIntoSchedule(ctx, courseIDs, itemIDs) + if err != nil { + return err + } + //批量更新任务块的embedded_time字段 + targetTimes := make([]*model.TargetTime, 0, len(toEmbed)) + for _, item := range toEmbed { + targetTimes = append(targetTimes, &model.TargetTime{ + DayOfWeek: item.DayOfWeek, + Week: item.Week, + SectionFrom: item.StartSection, + SectionTo: item.EndSection, + }) + } + err = txM.TaskClass.BatchUpdateTaskClassItemEmbeddedTime(ctx, itemIDs, targetTimes) + if err != nil { + return err + } + //5.2 再处理需要新建日程的任务块 + //先提取出需要新建日程的任务块ID列表 + normalItemIDs := make([]int, 0, len(toNormal)) + for _, item := range toNormal { + normalItemIDs = append(normalItemIDs, item.TaskItemID) + } + //验证一下plans中的taskItemID确实都属于这个任务类（避免用户恶意构造请求把别的用户的任务块或者不属于任何任务类的任务块也安排了） + result, err = sv.taskClassRepo.ValidateTaskItemIDsBelongToTaskClass(ctx, taskClassID, normalItemIDs) + if err != nil { + return err + } + if !result { + return respond.TaskClassItemNotBelongToTaskClass + } + //批量提取TaskItems + taskItems, err := txM.TaskClass.GetTaskClassItemsByIDs(ctx, normalItemIDs) + if err != nil { + return err + } + if len(taskItems) != len(normalItemIDs) { + log.Printf("警告：批量提取任务块时，返回的任务块数量与请求中的任务块ID数量不匹配，可能存在数据问题。请求ID数量：%d，返回任务块数量：%d", len(normalItemIDs), len(taskItems)) + return respond.InternalError(errors.New("返回的任务块数量与请求中的任务块ID数量不匹配，可能存在数据问题")) + } + //将toNormal按照TaskItemID升序排序，将taskItems也按照ID升序排序，保证一一对应关系（上面已经检查过重复） + //如果请求中的任务块ID有重复，这里就无法保证一一对应关系了，后续可以考虑在请求层面加一个校验，拒绝包含重复任务块ID的请求 + sort.SliceStable(toNormal, func(i, j int) bool { + return toNormal[i].TaskItemID < toNormal[j].TaskItemID + }) + sort.SliceStable(taskItems, func(i, j int) bool { + return taskItems[i].ID < taskItems[j].ID + }) + //开始构建event和schedules + finalSchedules := make([]model.Schedule, 0) //最终要插入数据库的Schedule切片 + finalScheduleEvents := make([]model.ScheduleEvent, 0) //最终要插入数据库的ScheduleEvent切片 + pos := make([]int, 0) //记录每个任务块对应的Schedule在finalSchedules中的位置，方便后续批量插入数据库后回填EventID + for i := 0; i < len(toNormal); i++ { + item := toNormal[i] + taskItem := taskItems[i] + if item.StartSection < 1 || item.EndSection > 12 || item.StartSection > item.EndSection { + return respond.InvalidSectionRange + } + schedules, scheduleEvent, err := conv.UserInsertTaskItemRequestToModel(&model.UserInsertTaskClassItemToScheduleRequest{ + Week: item.Week, + DayOfWeek: item.DayOfWeek, + StartSection: item.StartSection, + EndSection: item.EndSection, + EmbedCourseEventID: 0, //不嵌入课程 + }, &taskItem, nil, userID, item.StartSection, item.EndSection) + if err != nil { + return err + } + finalScheduleEvents = append(finalScheduleEvents, *scheduleEvent) + for range schedules { + pos = append(pos, len(finalScheduleEvents)-1) + } + finalSchedules = append(finalSchedules, schedules...) + } + //最后批量插入数据库 + //先插入ScheduleEvent表，获取生成的EventID，再批量插入Schedule表，最后批量更新TaskClassItem的embedded_time字段 + ids, err := txM.Schedule.InsertScheduleEvents(ctx, finalScheduleEvents) + if err != nil { + return err + } + // 将生成的 ScheduleEvent ID 赋值给对应的 Schedule 的 EventID 字段 + for i := range finalSchedules { + finalSchedules[i].EventID = ids[pos[i]] + } + if _, err = txM.Schedule.AddSchedules(finalSchedules); err != nil { + return err + } + //批量更新任务块的embedded_time字段 + targetTimes = make([]*model.TargetTime, 0, len(toEmbed)) + for _, item := range toNormal { + targetTimes = append(targetTimes, &model.TargetTime{ + DayOfWeek: item.DayOfWeek, + Week: item.Week, + SectionFrom: item.StartSection, + SectionTo: item.EndSection, + }) + } + //提取出所有需要更新的任务块ID + itemIDs = make([]int, 0, len(toNormal)) + for _, item := range toNormal { + itemIDs = append(itemIDs, item.TaskItemID) + } + err = txM.TaskClass.BatchUpdateTaskClassItemEmbeddedTime(ctx, itemIDs, targetTimes) + if err != nil { + return err + } + return nil + }); err != nil { + return err + } + return nil +} diff --git a/backend/shared/contracts/taskclass/types.go b/backend/shared/contracts/taskclass/types.go new file mode 100644 index 0000000..a50eb07 --- /dev/null +++ b/backend/shared/contracts/taskclass/types.go @@ -0,0 +1,88 @@ +package taskclass + +// TargetTime 表示任务块被安排到课表中的相对时间坐标。 +type TargetTime struct { + Week int `json:"week"` + DayOfWeek int `json:"day_of_week"` + SectionFrom int `json:"section_from"` + SectionTo int `json:"section_to"` +} + +// UpsertTaskClassRequest 是 task-class 服务新增/更新任务类的跨进程契约。 +// +// 职责边界： +// 1. UserID 由 gateway 鉴权后补齐，不信任前端传入值； +// 2. TaskClassID 仅 update 场景使用，add 场景保持 0； +// 3. 业务校验仍由 task-class 服务 sv 层统一执行。 +type UpsertTaskClassRequest struct { + UserID int `json:"user_id"` + TaskClassID int `json:"task_class_id,omitempty"` + Name string `json:"name" binding:"required"` + StartDate string `json:"start_date" binding:"required"` + EndDate string `json:"end_date" binding:"required"` + Mode string `json:"mode" binding:"required,oneof=auto manual"` + SubjectType string `json:"subject_type,omitempty"` + DifficultyLevel string `json:"difficulty_level,omitempty"` + CognitiveIntensity string `json:"cognitive_intensity,omitempty"` + Config UpsertTaskClassConfig `json:"config" binding:"required"` + Items []UpsertTaskClassItemConfig `json:"items" binding:"required"` +} + +type UpsertTaskClassConfig struct { + TotalSlots int `json:"total_slots" binding:"required,min=1"` + AllowFillerCourse bool `json:"allow_filler_course"` + Strategy string `json:"strategy" binding:"required,oneof=steady rapid"` + ExcludedSlots []int `json:"excluded_slots"` + ExcludedDaysOfWeek []int `json:"excluded_days_of_week"` +} + +type UpsertTaskClassItemConfig struct { + ID int `json:"id,omitempty"` + Order int `json:"order" binding:"required,min=1"` + Content string `json:"content" binding:"required"` + EmbeddedTime *TargetTime `json:"embedded_time"` +} + +type UserRequest struct { + UserID int `json:"user_id"` +} + +type GetTaskClassRequest struct { + UserID int `json:"user_id"` + TaskClassID int `json:"task_class_id"` +} + +type InsertTaskClassItemIntoScheduleRequest struct { + UserID int `json:"user_id"` + TaskItemID int `json:"task_item_id"` + Week int `json:"week" binding:"required,min=1"` + DayOfWeek int `json:"day_of_week" binding:"required,min=1,max=7"` + StartSection int `json:"start_section" binding:"required,min=1"` + EndSection int `json:"end_section" binding:"required,min=1,gtefield=StartSection"` + EmbedCourseEventID int `json:"embed_course_event_id"` +} + +type DeleteTaskClassItemRequest struct { + UserID int `json:"user_id"` + TaskItemID int `json:"task_item_id"` +} + +type DeleteTaskClassRequest struct { + UserID int `json:"user_id"` + TaskClassID int `json:"task_class_id"` +} + +type ApplyBatchIntoScheduleRequest struct { + UserID int `json:"user_id"` + TaskClassID int `json:"task_class_id" binding:"required"` + Items []SingleTaskClassItem `json:"items" binding:"required,dive,required"` +} + +type SingleTaskClassItem struct { + TaskItemID int `json:"task_item_id" binding:"required"` + Week int `json:"week" binding:"required,min=1"` + DayOfWeek int `json:"day_of_week" binding:"required,min=1,max=7"` + StartSection int `json:"start_section" binding:"required,min=1"` + EndSection int `json:"end_section" binding:"required,min=1,gtefield=StartSection"` + EmbedCourseEventID int `json:"embed_course_event_id"` +} diff --git a/backend/shared/ports/taskclass.go b/backend/shared/ports/taskclass.go new file mode 100644 index 0000000..76b8078 --- /dev/null +++ b/backend/shared/ports/taskclass.go @@ -0,0 +1,25 @@ +package ports + +import ( + "context" + "encoding/json" + + taskclasscontracts "github.com/LoveLosita/smartflow/backend/shared/contracts/taskclass" +) + +// TaskClassCommandClient 是 gateway 调用 task-class 服务的最小能力集合。 +// +// 职责边界： +// 1. 只覆盖当前 `/api/v1/task-class/*` HTTP 门面需要的能力； +// 2. 不暴露 task-class DAO、事务编排或迁移期 schedule 直写细节； +// 3. 复杂响应以 JSON 透传，避免 gateway 复制 task-class 内部 DTO。 +type TaskClassCommandClient interface { + AddTaskClass(ctx context.Context, req taskclasscontracts.UpsertTaskClassRequest) (json.RawMessage, error) + ListTaskClasses(ctx context.Context, userID int) (json.RawMessage, error) + GetTaskClass(ctx context.Context, req taskclasscontracts.GetTaskClassRequest) (json.RawMessage, error) + UpdateTaskClass(ctx context.Context, req taskclasscontracts.UpsertTaskClassRequest) (json.RawMessage, error) + InsertTaskClassItemIntoSchedule(ctx context.Context, req taskclasscontracts.InsertTaskClassItemIntoScheduleRequest) (json.RawMessage, error) + DeleteTaskClassItem(ctx context.Context, req taskclasscontracts.DeleteTaskClassItemRequest) (json.RawMessage, error) + DeleteTaskClass(ctx context.Context, req taskclasscontracts.DeleteTaskClassRequest) (json.RawMessage, error) + ApplyBatchIntoSchedule(ctx context.Context, req taskclasscontracts.ApplyBatchIntoScheduleRequest) (json.RawMessage, error) +} diff --git a/docs/backend/微服务四步迁移与第二阶段并行开发计划.md b/docs/backend/微服务四步迁移与第二阶段并行开发计划.md index 5aaef24..61de35b 100644 --- a/docs/backend/微服务四步迁移与第二阶段并行开发计划.md +++ b/docs/backend/微服务四步迁移与第二阶段并行开发计划.md @@ -430,7 +430,7 @@ flowchart LR ### 4.9 阶段 5：再拆 schedule / task / course / task-class -当前进展（2026-05-04）： +当前进展（2026-05-05）： 1. 首刀 `schedule` 已完成服务化：新增 `cmd/schedule`、`services/schedule/{dao,rpc,sv,core}`、`gateway/client/schedule`、`shared/contracts/schedule` 和 `shared/ports` schedule port。 2. gateway 的 `/api/v1/schedule/*` HTTP 门面已切到 schedule zrpc client；gateway 不再通过 `backend/service.ScheduleService` 直接承载 schedule HTTP 入口业务。 @@ -439,9 +439,13 @@ flowchart LR 5. gateway 的 `/api/v1/task/*` HTTP 门面已切到 task zrpc client；gateway 只负责鉴权、参数绑定、短超时和响应透传，不再直接调用 `backend/service.TaskService`。 6. active-scheduler 的 task facts / due job scanner 已切到 task RPC adapter；`cmd/active-scheduler` 启动依赖检查已移除 `tasks`，进一步缩小 active-scheduler 对跨域主库表的直接依赖。 7. `task.urgency.promote.requested` 的 handler、relay、retry loop 已迁入 `cmd/task`；单体 outbox worker 只保留 agent / memory consumer，Agent 残留查询链路只允许 publish-only 写入 `task_outbox_messages`，避免单体和 task 独立服务抢同一 task consumer group。 -8. 旧实现仍保留：`backend/service/schedule.go`、`backend/dao/schedule.go`、`backend/service/task.go`、`backend/dao/task.go`、active-scheduler 旧 Gorm apply adapter 暂时保留，用于 agent 迁移期、单体残留路径和回退。 -9. 当前切流点：HTTP schedule 流量进入 `cmd/schedule`；HTTP task 流量进入 `cmd/task`；active-scheduler 读取 task/schedule facts 与正式写日程均走 RPC；course / task-class / agent 内部仍存在直接 DAO 调用，后续按域继续切。 -10. 当前残留跨域 DB 依赖：task 服务迁移期仍 best-effort 写 `active_schedule_jobs`；active-scheduler 仍直接写 agent 会话 / timeline 和 notification outbox 相关表；agent 本地 task 查询链路仍保留旧 `TaskService` 作为迁移期适配。 +8. 第三刀 `task-class` 已完成 HTTP 所有权切流：新增 `cmd/task-class`、`services/task_class/{dao,rpc,sv}`、`gateway/client/taskclass`、`shared/contracts/taskclass` 和 `shared/ports` task-class port。 +9. gateway 的 `/api/v1/task-class/*` HTTP 门面已切到 task-class zrpc client；gateway 只负责鉴权、参数绑定、短超时和响应透传，不再直接调用 `backend/service.TaskClassService`。 +10. task-class 本轮按主人拍板保留迁移期直写 `schedule_events` / `schedules` 权限，不走 schedule RPC bridge，以保留 `insert into schedule` / `apply batch into schedule` 与 task item 状态更新的本地事务语义；`cmd/task-class` 只 AutoMigrate `task_classes` / `task_items`，启动时显式检查 schedule 依赖表是否存在。 +11. 旧实现仍保留：`backend/service/schedule.go`、`backend/dao/schedule.go`、`backend/service/task.go`、`backend/dao/task.go`、`backend/service/task-class.go`、`backend/dao/task-class.go`、active-scheduler 旧 Gorm apply adapter 暂时保留，用于 agent 迁移期、单体残留路径和回退。 +12. 当前切流点：HTTP schedule 流量进入 `cmd/schedule`；HTTP task 流量进入 `cmd/task`；HTTP task-class 流量进入 `cmd/task-class`；active-scheduler 读取 task/schedule facts 与正式写日程均走 RPC；course / agent 内部仍存在直接 DAO 调用，后续按域继续切。 +13. 当前残留跨域 DB 依赖：task-class 迁移期仍直接写 `schedule_events` / `schedules`；task 服务迁移期仍 best-effort 写 `active_schedule_jobs`；active-scheduler 仍直接写 agent 会话 / timeline 和 notification outbox 相关表；agent 本地 task 查询、task-class upsert 和 schedule provider 仍保留 DAO 适配。 +14. 已完成验证：`go test ./...` 通过；避让默认端口启动完整本地服务组（HTTP `18080`，zrpc `19081-19086`）后，task-class add / list / get / insert-into-schedule / delete-item / delete-class smoke 通过，并用 `docker exec` 核对 task-class 与 schedule 相关表无残留。 目标： @@ -452,7 +456,7 @@ flowchart LR 这一步要做的事： -1. `schedule`、`task` 已先后独立；下一轮优先评估 `task-class`，再看 `course`。 +1. `schedule`、`task`、`task-class` 已先后独立；下一轮优先评估 `course`，再看 agent 残留 DAO 适配。 2. 每个领域只维护自己的写模型。 3. 通过事件或明确 RPC 契约通信。 4. 继续保持并行迁移，旧实现和新实现可以短期并存。 @@ -461,7 +465,8 @@ flowchart LR 1. schedule 切流完成后 commit。 2. task 切流完成后 commit。 -3. course / task-class 切流完成后 commit。 +3. task-class 切流完成后 commit。 +4. course 切流完成后 commit。 建议测试： @@ -894,7 +899,7 @@ graph TD 6. 阶段 4 `active-scheduler` 已完成首轮收口；后续不要再把它当成“未拆服务”，除非是在补契约测试或继续替换跨域 DB 访问。 7. `shared` 只保留跨进程契约和少量跨服务底座，不承载业务逻辑、DAO、模型或状态机。 8. 如果后续要改目录，必须先回答“这个文件属于哪一个典型用例”，回答不清楚就先别动结构。 -9. 当前文档已经可以作为切对话基线；后续代理默认按本文件推进。现阶段的迁移基线入口是 `backend/cmd/api`、`backend/cmd/worker`、`backend/cmd/all`，它们只是当前仓库的启动壳，不是终态。`backend/cmd/userauth` 是阶段 2 的独立服务入口，`backend/cmd/notification` 是阶段 3 的独立服务入口，`backend/cmd/active-scheduler` 是阶段 4 的独立服务入口。终态仍然是“一个服务一个独立 `main.go`”，只在出现新的契约风险、边界变化或业务语义变化时再重新讨论架构。 +9. 当前文档已经可以作为切对话基线；后续代理默认按本文件推进。现阶段的迁移基线入口是 `backend/cmd/api`、`backend/cmd/worker`、`backend/cmd/all`，它们只是当前仓库的启动壳，不是终态。`backend/cmd/userauth` 是阶段 2 的独立服务入口，`backend/cmd/notification` 是阶段 3 的独立服务入口，`backend/cmd/active-scheduler` 是阶段 4 的独立服务入口，`backend/cmd/schedule`、`backend/cmd/task`、`backend/cmd/task-class` 是阶段 5 已落地的独立服务入口。终态仍然是“一个服务一个独立 `main.go`”，只在出现新的契约风险、边界变化或业务语义变化时再重新讨论架构。 ### 6.10 启动方式与进程模型 @@ -905,6 +910,7 @@ graph TD 5. 如果某些服务需要联动启动，应通过脚本、Makefile、docker compose 或开发编排器去启动多个二进制，而不是把进程边界打穿。 6. 带 worker 的服务可以继续保留多入口角色，例如 `api` / `worker` / `all`，但它们仍然是同一服务的不同可执行角色，不是把多个服务硬塞进一个进程。 7. MySQL / Redis 容器的启动归 `docker compose` 或运维层；Go 服务只负责在自己的进程里建立连接、做自己的 AutoMigrate 和连通性检查。 +8. 阶段 5 后，旧 `cmd/start.go` / `cmd/all` 只是 gateway 和迁移期组合壳；本地完整 smoke 必须额外启动 `cmd/userauth`、`cmd/notification`、`cmd/active-scheduler`、`cmd/schedule`、`cmd/task` 和 `cmd/task-class`。如果同机已有另一条线占用默认端口，应复制临时配置，把 HTTP / zrpc 端口整体平移后再启动服务。 ### 6.11 测试自动化与 smoke 权限边界 @@ -1088,13 +1094,14 @@ graph TD 阶段 5 当前基线： -1. `backend/cmd/schedule/main.go` 是 schedule 独立进程入口，`backend/cmd/task/main.go` 是 task 独立进程入口，二者各自初始化 DB / Redis / zrpc server 和所需服务内资源。 -2. `backend/services/schedule` 拥有正式日程领域核心，`backend/services/task` 拥有任务池读写、完成/撤销、紧急性平移和 task outbox handler。 -3. `backend/gateway/api` 继续作为 HTTP 门面统一目录，`backend/gateway/client/schedule` 与 `backend/gateway/client/task` 作为 gateway 侧 zrpc client。 -4. `backend/shared/contracts/schedule`、`backend/shared/contracts/task` 和 `backend/shared/ports` 只承载跨进程契约与端口接口，不放 DAO、model 或业务状态机。 +1. `backend/cmd/schedule/main.go` 是 schedule 独立进程入口，`backend/cmd/task/main.go` 是 task 独立进程入口，`backend/cmd/task-class/main.go` 是 task-class 独立进程入口，三者各自初始化 DB / Redis / zrpc server 和所需服务内资源。 +2. `backend/services/schedule` 拥有正式日程领域核心，`backend/services/task` 拥有任务池读写、完成/撤销、紧急性平移和 task outbox handler，`backend/services/task_class` 拥有任务类与任务块维护、批量排入日程等核心逻辑。 +3. `backend/gateway/api` 继续作为 HTTP 门面统一目录，`backend/gateway/client/schedule`、`backend/gateway/client/task` 与 `backend/gateway/client/taskclass` 作为 gateway 侧 zrpc client。 +4. `backend/shared/contracts/schedule`、`backend/shared/contracts/task`、`backend/shared/contracts/taskclass` 和 `backend/shared/ports` 只承载跨进程契约与端口接口，不放 DAO、model 或业务状态机。 5. active-scheduler 的 schedule facts / feedback / confirm apply 已走 schedule RPC，task facts / due job scanner 已走 task RPC；启动依赖检查不再要求 `schedule_events`、`schedules`、`task_classes`、`task_items` 或 `tasks`。 6. `task.urgency.promote.requested` 的消费边界已迁入 `cmd/task`；单体 outbox worker 不再启动 task service bus，只保留 Agent 残留路径的 publish-only 写入能力，避免迁移期重复 relay / consume。 -7. 本阶段残留：task 服务仍 best-effort 写 `active_schedule_jobs`；agent 本地 task 查询、quick task 创建、course / task-class 仍存在直接 DAO 调用；active-scheduler 旧 Gorm apply adapter 保留为迁移期残留，不作为新流量主路径。 +7. task-class 迁移期仍直接写 `schedule_events` / `schedules`，用于保留 `insert/apply` 与 item 状态更新的本地事务语义；服务启动只检查这些 schedule 依赖表，不 AutoMigrate schedule 表。 +8. 本阶段残留：task 服务仍 best-effort 写 `active_schedule_jobs`；agent 本地 task 查询、quick task 创建、task-class upsert、schedule provider 和 course 仍存在直接 DAO 调用；active-scheduler 旧 Gorm apply adapter 保留为迁移期残留，不作为新流量主路径。 ---