From 7d21b6516f9a53355311eb6effbfce79b19943d0 Mon Sep 17 00:00:00 2001 From: Losita <2810873701@qq.com> Date: Wed, 29 Apr 2026 17:44:42 +0800 Subject: [PATCH] Version: 0.9.56.dev.260429 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后端： 1. 启动层完成第一轮运行边界拆分，新增 `all / api / worker` 三种进程模式：`all` 保持原单体行为，`api` 只启动 Gin 与同步业务依赖，`worker` 只启动 outbox、Kafka consumer 与 memory worker。 2. 启动装配从单个入口拆成 runtime 依赖图，配置、DB、Redis、RAG、memory、DAO、Service、Handler、newAgent 依赖统一集中构造，再按进程角色选择启动 HTTP 或后台循环。 3. outbox 事件总线补齐 dispatch / consume 分离启动能力，支持后续 relay 与 consumer 独立进程化，同时保留原组合启动语义。 4. 核心 outbox handler 注册收口为公共接线入口，统一校验依赖并复用注册顺序，避免 api / worker / all 多入口复制事件注册逻辑。 迁移说明： 5. 本轮只迁运行边界，不拆业务服务边界；旧单体入口仍保留并默认走 `all` 兼容模式，当前切流点是 API 不再消费异步事件，worker 承担后台消费与 memory 任务。 6. 补充微服务四步迁移与第二阶段并行开发计划，明确先拆 API/Worker，再接主动调度与飞书通知，后续再拆 notification、active-scheduler、schedule/task。 --- backend/cmd/all/main.go | 7 + backend/cmd/api/main.go | 7 + backend/cmd/start.go | 580 ++++++++++++------ backend/cmd/worker/main.go | 7 + backend/infra/outbox/engine.go | 26 + backend/infra/outbox/event_bus.go | 26 + .../service/events/core_outbox_handlers.go | 87 +++ .../微服务四步迁移与第二阶段并行开发计划.md | 557 +++++++++++++++++ 8 files changed, 1097 insertions(+), 200 deletions(-) create mode 100644 backend/cmd/all/main.go create mode 100644 backend/cmd/api/main.go create mode 100644 backend/cmd/worker/main.go create mode 100644 backend/service/events/core_outbox_handlers.go create mode 100644 docs/backend/微服务四步迁移与第二阶段并行开发计划.md diff --git a/backend/cmd/all/main.go b/backend/cmd/all/main.go new file mode 100644 index 0000000..ec80bf5 --- /dev/null +++ b/backend/cmd/all/main.go @@ -0,0 +1,7 @@ +package main + +import "github.com/LoveLosita/smartflow/backend/cmd" + +func main() { + cmd.StartAll() +} diff --git a/backend/cmd/api/main.go b/backend/cmd/api/main.go new file mode 100644 index 0000000..62119a8 --- /dev/null +++ b/backend/cmd/api/main.go @@ -0,0 +1,7 @@ +package main + +import "github.com/LoveLosita/smartflow/backend/cmd" + +func main() { + cmd.StartAPI() +} diff --git a/backend/cmd/start.go b/backend/cmd/start.go index f5542f7..7a3f863 100644 --- a/backend/cmd/start.go +++ b/backend/cmd/start.go @@ -5,7 +5,9 @@ import ( "fmt" "log" "os" + "os/signal" "strings" + "syscall" "time" "github.com/LoveLosita/smartflow/backend/api" @@ -17,6 +19,7 @@ import ( ragconfig "github.com/LoveLosita/smartflow/backend/infra/rag/config" "github.com/LoveLosita/smartflow/backend/inits" "github.com/LoveLosita/smartflow/backend/memory" + memorymodel "github.com/LoveLosita/smartflow/backend/memory/model" memoryobserve "github.com/LoveLosita/smartflow/backend/memory/observe" "github.com/LoveLosita/smartflow/backend/middleware" "github.com/LoveLosita/smartflow/backend/model" @@ -28,14 +31,40 @@ import ( "github.com/LoveLosita/smartflow/backend/routers" "github.com/LoveLosita/smartflow/backend/service" eventsvc "github.com/LoveLosita/smartflow/backend/service/events" + "github.com/go-redis/redis/v8" "github.com/spf13/viper" + "gorm.io/gorm" ) +// appRuntime 承载一次进程启动所需的依赖图。 +// +// 职责边界： +// 1. 只负责保存启动期已经装配好的基础设施、仓储、服务和 HTTP handler； +// 2. 不承载业务逻辑，业务仍然由 service / newAgent / memory 等领域模块负责； +// 3. 不决定进程角色，api / worker / all 由 StartAPI、StartWorker、StartAll 选择启动哪些生命周期。 +type appRuntime struct { + db *gorm.DB + redisClient *redis.Client + cacheRepo *dao.CacheDAO + userRepo *dao.UserDAO + agentRepo *dao.AgentDAO + agentCache *dao.AgentCache + manager *dao.RepoManager + outboxRepo *outboxinfra.Repository + eventBus *outboxinfra.EventBus + memoryModule *memory.Module + limiter *pkg.RateLimiter + handlers *api.ApiHandlers +} + // loadConfig 加载应用配置。 func loadConfig() error { viper.SetConfigName("config") viper.SetConfigType("yaml") viper.AddConfigPath(".") + // 1. 兼容从仓库根目录执行 `go run ./backend/cmd/api` 的场景； + // 2. 从 backend 目录执行时仍优先命中当前目录，不改变现有默认行为。 + viper.AddConfigPath("backend") if err := viper.ReadInConfig(); err != nil { return fmt.Errorf("failed to read config file: %w", err) } @@ -43,15 +72,82 @@ func loadConfig() error { return nil } -// Start 是应用启动入口。 +// Start 保留历史入口，默认仍按 all 模式启动。 +// +// 职责边界： +// 1. 兼容 backend/main.go 以及旧部署命令； +// 2. 不新增业务语义，只委托给 StartAll； +// 3. 后续若部署全面切到独立 api/worker，可逐步废弃该兼容入口。 func Start() { + StartAll() +} + +// StartAll 启动迁移期兼容模式：HTTP API 与后台 worker 在同一进程内运行。 +func StartAll() { + ctx := context.Background() + runtime := mustBuildRuntime(ctx) + defer runtime.close() + + runtime.startWorkers(ctx) + runtime.startHTTP() +} + +// StartAPI 只启动 Gin API 及其同步 service/dao 依赖，不启动后台 worker。 +// +// 说明： +// 1. 该模式仍是“带 service/dao 的 API 单体”，不是最终 API Gateway； +// 2. API 可以继续写入 outbox，但不负责消费 outbox，也不启动 memory worker； +// 3. worker 停止时，API 仍可提供同步接口，只是异步能力会延迟处理。 +func StartAPI() { + ctx := context.Background() + runtime := mustBuildRuntime(ctx) + defer runtime.close() + + runtime.startHTTP() +} + +// StartWorker 只启动后台异步能力，不注册 Gin 路由。 +// +// 运行内容： +// 1. outbox relay：扫描 pending 消息并投递 Kafka； +// 2. Kafka consumer：消费事件并分发到业务 handler； +// 3. memory worker：处理 memory_jobs 后台任务。 +func StartWorker() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + runtime := mustBuildRuntime(ctx) + defer runtime.close() + + runtime.startWorkers(ctx) + log.Println("Worker process started") + <-ctx.Done() + log.Println("Worker process stopping") +} + +func mustBuildRuntime(ctx context.Context) *appRuntime { + runtime, err := buildRuntime(ctx) + if err != nil { + log.Fatalf("Failed to initialize application runtime: %v", err) + } + return runtime +} + +// buildRuntime 装配应用依赖图，但不启动 HTTP 或后台循环。 +// +// 步骤说明： +// 1. 先初始化配置、数据库、Redis、模型、RAG、memory 等基础设施； +// 2. 再构造 DAO / Service / newAgent 依赖； +// 3. 最后构造 HTTP handlers，供 api/all 模式按需启动； +// 4. worker 模式暂时也复用完整依赖图，避免同轮迁移拆出两套装配逻辑。 +func buildRuntime(ctx context.Context) (*appRuntime, error) { if err := loadConfig(); err != nil { - log.Fatalf("Failed to load config: %v", err) + return nil, err } db, err := inits.ConnectDB() if err != nil { - log.Fatalf("Failed to connect to database: %v", err) + return nil, fmt.Errorf("failed to connect to database: %w", err) } rdb := inits.InitRedis() @@ -59,30 +155,14 @@ func Start() { aiHub, err := inits.InitEino() if err != nil { - log.Fatalf("Failed to initialize Eino: %v", err) + return nil, fmt.Errorf("failed to initialize Eino: %w", err) } - ragCfg := ragconfig.LoadFromViper() - var ragRuntime infrarag.Runtime - if ragCfg.Enabled { - // 1. 当前项目尚未完成全局观测平台建设，这里先注入一层轻量 Observer； - // 2. RAG 内部只依赖 Observer 接口，后续若全项目统一日志/指标系统，只需替换这里； - // 3. 这样可以避免 RAG 单独自建一套割裂的日志基础设施。 - ragLogger := log.Default() - ragRuntime, err = infrarag.NewRuntimeFromConfig(context.Background(), ragCfg, infrarag.FactoryDeps{ - Logger: ragLogger, - Observer: infrarag.NewLoggerObserver(ragLogger), - }) - if err != nil { - log.Fatalf("Failed to initialize RAG runtime: %v", err) - } - log.Printf("RAG runtime initialized: store=%s embed=%s reranker=%s", ragCfg.Store, ragCfg.EmbedProvider, ragCfg.RerankerProvider) - } else { - log.Println("RAG runtime is disabled") + ragRuntime, err := buildRAGRuntime(ctx) + if err != nil { + return nil, err } - // 1. memory 模块对启动层只暴露一个门面。 - // 2. 后续若接入统一 DI 容器，也优先注入这个门面，而不是继续暴露内部 repo/service。 memoryCfg := memory.LoadConfigFromViper() memoryObserver := memoryobserve.NewLoggerObserver(log.Default()) memoryMetrics := memoryobserve.NewMetricsRegistry() @@ -110,55 +190,94 @@ func Start() { agentRepo := dao.NewAgentDAO(db) outboxRepo := outboxinfra.NewRepository(db) - // outbox 通用事件总线接线（第二阶段）： - // 1. 读取 Kafka 配置； - // 2. 创建 infra 级 EventBus； - // 3. 显式注册业务事件处理器； - // 4. 启动总线后台 dispatch/consume 循环。 - kafkaCfg := kafkabus.LoadConfig() - eventBus, err := outboxinfra.NewEventBus(outboxRepo, kafkaCfg) + eventBus, err := buildEventBus(outboxRepo) if err != nil { - log.Fatalf("Failed to initialize outbox event bus: %v", err) + return nil, err } - if eventBus != nil { - // 1. 在启动前完成业务事件处理器注册。 - // 2. memory 事件处理器也统一通过 memoryModule 接入，避免启动层感知内部细节。 - if err = eventsvc.RegisterChatHistoryPersistHandler(eventBus, outboxRepo, manager); err != nil { - log.Fatalf("Failed to register chat history event handler: %v", err) - } - if err = eventsvc.RegisterTaskUrgencyPromoteHandler(eventBus, outboxRepo, manager); err != nil { - log.Fatalf("Failed to register task urgency promote event handler: %v", err) - } - if err = eventsvc.RegisterChatTokenUsageAdjustHandler(eventBus, outboxRepo, manager); err != nil { - log.Fatalf("Failed to register chat token usage adjust event handler: %v", err) - } - if err = eventsvc.RegisterAgentStateSnapshotHandler(eventBus, outboxRepo, manager); err != nil { - log.Fatalf("Failed to register agent state snapshot event handler: %v", err) - } - if err = eventsvc.RegisterAgentTimelinePersistHandler(eventBus, outboxRepo, agentRepo, cacheRepo); err != nil { - log.Fatalf("Failed to register agent timeline persist event handler: %v", err) - } - if err = eventsvc.RegisterMemoryExtractRequestedHandler(eventBus, outboxRepo, memoryModule); err != nil { - log.Fatalf("Failed to register memory extract event handler: %v", err) - } - eventBus.Start(context.Background()) - defer eventBus.Close() - log.Println("Outbox event bus started") - } else { - log.Println("Outbox event bus is disabled") - } - - memoryModule.StartWorker(context.Background()) // Service 层初始化。 userService := service.NewUserService(userRepo, cacheRepo) taskSv := service.NewTaskService(taskRepo, cacheRepo, eventBus) + courseService := buildCourseService(courseRepo, scheduleRepo) + taskClassService := service.NewTaskClassService(taskClassRepo, cacheRepo, scheduleRepo, manager) + scheduleService := service.NewScheduleService(scheduleRepo, userRepo, taskClassRepo, manager, cacheRepo) + agentService := service.NewAgentServiceWithSchedule(aiHub, agentRepo, taskRepo, cacheRepo, agentCacheRepo, eventBus, scheduleService, taskSv) + + configureAgentService( + agentService, + ragRuntime, + agentRepo, + cacheRepo, + taskRepo, + taskClassRepo, + scheduleRepo, + memoryModule, + memoryCfg, + ) + + handlers := buildAPIHandlers(userService, taskSv, taskClassService, courseService, scheduleService, agentService, memoryModule) + + return &appRuntime{ + db: db, + redisClient: rdb, + cacheRepo: cacheRepo, + userRepo: userRepo, + agentRepo: agentRepo, + agentCache: agentCacheRepo, + manager: manager, + outboxRepo: outboxRepo, + eventBus: eventBus, + memoryModule: memoryModule, + limiter: limiter, + handlers: handlers, + }, nil +} + +func buildRAGRuntime(ctx context.Context) (infrarag.Runtime, error) { + ragCfg := ragconfig.LoadFromViper() + if !ragCfg.Enabled { + log.Println("RAG runtime is disabled") + return nil, nil + } + + // 1. 当前项目尚未完成全局观测平台建设，这里先注入一层轻量 Observer； + // 2. RAG 内部只依赖 Observer 接口，后续若全项目统一日志/指标系统，只需替换这里； + // 3. 这样可以避免 RAG 单独自建一套割裂的日志基础设施。 + ragLogger := log.Default() + ragRuntime, err := infrarag.NewRuntimeFromConfig(ctx, ragCfg, infrarag.FactoryDeps{ + Logger: ragLogger, + Observer: infrarag.NewLoggerObserver(ragLogger), + }) + if err != nil { + return nil, fmt.Errorf("failed to initialize RAG runtime: %w", err) + } + log.Printf("RAG runtime initialized: store=%s embed=%s reranker=%s", ragCfg.Store, ragCfg.EmbedProvider, ragCfg.RerankerProvider) + return ragRuntime, nil +} + +func buildEventBus(outboxRepo *outboxinfra.Repository) (*outboxinfra.EventBus, error) { + // outbox 通用事件总线接线： + // 1. API 模式只使用 Publish 写入 outbox，不启动后台循环； + // 2. worker/all 模式再显式注册 handler 并启动后台循环； + // 3. kafka.enabled=false 时返回 nil，业务按既有降级策略执行。 + kafkaCfg := kafkabus.LoadConfig() + eventBus, err := outboxinfra.NewEventBus(outboxRepo, kafkaCfg) + if err != nil { + return nil, fmt.Errorf("failed to initialize outbox event bus: %w", err) + } + if eventBus == nil { + log.Println("Outbox event bus is disabled") + } + return eventBus, nil +} + +func buildCourseService(courseRepo *dao.CourseDAO, scheduleRepo *dao.ScheduleDAO) *service.CourseService { courseImageResponsesClient := infrallm.NewArkResponsesClient( os.Getenv("ARK_API_KEY"), viper.GetString("agent.baseURL"), viper.GetString("courseImport.visionModel"), ) - courseService := service.NewCourseService( + return service.NewCourseService( courseRepo, scheduleRepo, courseImageResponsesClient, @@ -168,15 +287,26 @@ func Start() { ), viper.GetString("courseImport.visionModel"), ) - taskClassService := service.NewTaskClassService(taskClassRepo, cacheRepo, scheduleRepo, manager) - scheduleService := service.NewScheduleService(scheduleRepo, userRepo, taskClassRepo, manager, cacheRepo) - agentService := service.NewAgentServiceWithSchedule(aiHub, agentRepo, taskRepo, cacheRepo, agentCacheRepo, eventBus, scheduleService, taskSv) +} + +func configureAgentService( + agentService *service.AgentService, + ragRuntime infrarag.Runtime, + agentRepo *dao.AgentDAO, + cacheRepo *dao.CacheDAO, + taskRepo *dao.TaskDAO, + taskClassRepo *dao.TaskClassDAO, + scheduleRepo *dao.ScheduleDAO, + memoryModule *memory.Module, + memoryCfg memorymodel.Config, +) { + if agentService == nil { + return + } // newAgent 依赖接线。 agentService.SetAgentStateStore(dao.NewAgentStateStoreAdapter(cacheRepo)) - // 1. WebSearch provider 初始化：根据配置选择 mock/bocha； - // 2. provider 为 nil 时，web_search / web_fetch 返回"暂未启用"，不阻断主流程。 var webSearchProvider web.SearchProvider webProvider := viper.GetString("websearch.provider") switch webProvider { @@ -202,150 +332,200 @@ func Start() { RAGRuntime: ragRuntime, WebSearchProvider: webSearchProvider, TaskClassWriteDeps: newagenttools.TaskClassWriteDeps{ - UpsertTaskClass: func(userID int, input newagenttools.TaskClassUpsertInput) (newagenttools.TaskClassUpsertPersistResult, error) { - req := input.Request - taskClassID := 0 - created := input.ID == 0 - - err := taskClassRepo.Transaction(func(txDAO *dao.TaskClassDAO) error { - // 1. 先构造任务类主体，保持与现有 AddOrUpdateTaskClass 口径一致。 - taskClass := &model.TaskClass{ - ID: input.ID, - Name: &req.Name, - Mode: &req.Mode, - SubjectType: stringPtrOrNil(req.SubjectType), - DifficultyLevel: stringPtrOrNil(req.DifficultyLevel), - CognitiveIntensity: stringPtrOrNil(req.CognitiveIntensity), - TotalSlots: &req.Config.TotalSlots, - Strategy: &req.Config.Strategy, - ExcludedSlots: req.Config.ExcludedSlots, - ExcludedDaysOfWeek: req.Config.ExcludedDaysOfWeek, - } - taskClass.AllowFillerCourse = &req.Config.AllowFillerCourse - - // 2. 自动模式下写入日期范围；手动模式允许为空。 - if req.StartDate != "" { - startDate, parseErr := time.ParseInLocation("2006-01-02", req.StartDate, time.Local) - if parseErr != nil { - return parseErr - } - taskClass.StartDate = &startDate - } - if req.EndDate != "" { - endDate, parseErr := time.ParseInLocation("2006-01-02", req.EndDate, time.Local) - if parseErr != nil { - return parseErr - } - taskClass.EndDate = &endDate - } - - // 3. upsert 主体后拿到稳定 task_class_id，供 items 绑定 category_id。 - updatedID, upsertErr := txDAO.AddOrUpdateTaskClass(userID, taskClass) - if upsertErr != nil { - return upsertErr - } - taskClassID = updatedID - - // 4. 构造任务块并批量 upsert。 - items := make([]model.TaskClassItem, 0, len(req.Items)) - for _, itemReq := range req.Items { - categoryID := taskClassID - order := itemReq.Order - content := itemReq.Content - status := model.TaskItemStatusUnscheduled - items = append(items, model.TaskClassItem{ - ID: itemReq.ID, - CategoryID: &categoryID, - Order: &order, - Content: &content, - EmbeddedTime: itemReq.EmbeddedTime, - Status: &status, - }) - } - return txDAO.AddOrUpdateTaskClassItems(userID, items) - }) - if err != nil { - return newagenttools.TaskClassUpsertPersistResult{}, err - } - return newagenttools.TaskClassUpsertPersistResult{ - TaskClassID: taskClassID, - Created: created, - }, nil - }, + UpsertTaskClass: buildTaskClassUpsertFunc(taskClassRepo), }, })) agentService.SetScheduleProvider(newagentconv.NewScheduleProvider(scheduleRepo, taskClassRepo)) agentService.SetCompactionStore(agentRepo) agentService.SetQuickTaskDeps(newagentmodel.QuickTaskDeps{ - CreateTask: func(userID int, title string, priorityGroup int, deadlineAt *time.Time, urgencyThresholdAt *time.Time) (int, error) { - created, err := taskRepo.AddTask(&model.Task{ - UserID: userID, - Title: title, - Priority: priorityGroup, - IsCompleted: false, - DeadlineAt: deadlineAt, - UrgencyThresholdAt: urgencyThresholdAt, - }) - if err != nil { - return 0, err - } - return created.ID, nil - }, - QueryTasks: func(ctx context.Context, userID int, params newagentmodel.TaskQueryParams) ([]newagentmodel.TaskQueryResult, error) { - req := newagentmodel.TaskQueryRequest{ - UserID: userID, - Quadrant: params.Quadrant, - SortBy: params.SortBy, - Order: params.Order, - Limit: params.Limit, - IncludeCompleted: params.IncludeCompleted, - Keyword: params.Keyword, - DeadlineBefore: params.DeadlineBefore, - DeadlineAfter: params.DeadlineAfter, - } - records, err := agentService.QueryTasksForTool(ctx, req) - if err != nil { - return nil, err - } - results := make([]newagentmodel.TaskQueryResult, 0, len(records)) - for _, r := range records { - deadlineStr := "" - if r.DeadlineAt != nil { - deadlineStr = r.DeadlineAt.In(time.Local).Format("2006-01-02 15:04") - } - results = append(results, newagentmodel.TaskQueryResult{ - ID: r.ID, - Title: r.Title, - PriorityGroup: r.PriorityGroup, - IsCompleted: r.IsCompleted, - DeadlineAt: deadlineStr, - }) - } - return results, nil - }, + CreateTask: buildQuickTaskCreateFunc(taskRepo), + QueryTasks: buildQuickTaskQueryFunc(agentService), }) agentService.SetMemoryReader(memoryModule, memoryCfg) +} - // API 层初始化。 - userApi := api.NewUserHandler(userService) - taskApi := api.NewTaskHandler(taskSv) - courseApi := api.NewCourseHandler(courseService) - taskClassApi := api.NewTaskClassHandler(taskClassService) - scheduleApi := api.NewScheduleAPI(scheduleService) - agentApi := api.NewAgentHandler(agentService) - memoryApi := api.NewMemoryHandler(memoryModule) - handlers := &api.ApiHandlers{ - UserHandler: userApi, - TaskHandler: taskApi, - TaskClassHandler: taskClassApi, - CourseHandler: courseApi, - ScheduleHandler: scheduleApi, - AgentHandler: agentApi, - MemoryHandler: memoryApi, +func buildTaskClassUpsertFunc(taskClassRepo *dao.TaskClassDAO) func(userID int, input newagenttools.TaskClassUpsertInput) (newagenttools.TaskClassUpsertPersistResult, error) { + return func(userID int, input newagenttools.TaskClassUpsertInput) (newagenttools.TaskClassUpsertPersistResult, error) { + req := input.Request + taskClassID := 0 + created := input.ID == 0 + + err := taskClassRepo.Transaction(func(txDAO *dao.TaskClassDAO) error { + // 1. 先构造任务类主体，保持与现有 AddOrUpdateTaskClass 口径一致。 + taskClass := &model.TaskClass{ + ID: input.ID, + Name: &req.Name, + Mode: &req.Mode, + SubjectType: stringPtrOrNil(req.SubjectType), + DifficultyLevel: stringPtrOrNil(req.DifficultyLevel), + CognitiveIntensity: stringPtrOrNil(req.CognitiveIntensity), + TotalSlots: &req.Config.TotalSlots, + Strategy: &req.Config.Strategy, + ExcludedSlots: req.Config.ExcludedSlots, + ExcludedDaysOfWeek: req.Config.ExcludedDaysOfWeek, + } + taskClass.AllowFillerCourse = &req.Config.AllowFillerCourse + + // 2. 自动模式下写入日期范围；手动模式允许为空。 + if req.StartDate != "" { + startDate, parseErr := time.ParseInLocation("2006-01-02", req.StartDate, time.Local) + if parseErr != nil { + return parseErr + } + taskClass.StartDate = &startDate + } + if req.EndDate != "" { + endDate, parseErr := time.ParseInLocation("2006-01-02", req.EndDate, time.Local) + if parseErr != nil { + return parseErr + } + taskClass.EndDate = &endDate + } + + // 3. upsert 主体后拿到稳定 task_class_id，供 items 绑定 category_id。 + updatedID, upsertErr := txDAO.AddOrUpdateTaskClass(userID, taskClass) + if upsertErr != nil { + return upsertErr + } + taskClassID = updatedID + + // 4. 构造任务块并批量 upsert。 + items := make([]model.TaskClassItem, 0, len(req.Items)) + for _, itemReq := range req.Items { + categoryID := taskClassID + order := itemReq.Order + content := itemReq.Content + status := model.TaskItemStatusUnscheduled + items = append(items, model.TaskClassItem{ + ID: itemReq.ID, + CategoryID: &categoryID, + Order: &order, + Content: &content, + EmbeddedTime: itemReq.EmbeddedTime, + Status: &status, + }) + } + return txDAO.AddOrUpdateTaskClassItems(userID, items) + }) + if err != nil { + return newagenttools.TaskClassUpsertPersistResult{}, err + } + return newagenttools.TaskClassUpsertPersistResult{ + TaskClassID: taskClassID, + Created: created, + }, nil + } +} + +func buildQuickTaskCreateFunc(taskRepo *dao.TaskDAO) func(userID int, title string, priorityGroup int, deadlineAt *time.Time, urgencyThresholdAt *time.Time) (int, error) { + return func(userID int, title string, priorityGroup int, deadlineAt *time.Time, urgencyThresholdAt *time.Time) (int, error) { + created, err := taskRepo.AddTask(&model.Task{ + UserID: userID, + Title: title, + Priority: priorityGroup, + IsCompleted: false, + DeadlineAt: deadlineAt, + UrgencyThresholdAt: urgencyThresholdAt, + }) + if err != nil { + return 0, err + } + return created.ID, nil + } +} + +func buildQuickTaskQueryFunc(agentService *service.AgentService) func(ctx context.Context, userID int, params newagentmodel.TaskQueryParams) ([]newagentmodel.TaskQueryResult, error) { + return func(ctx context.Context, userID int, params newagentmodel.TaskQueryParams) ([]newagentmodel.TaskQueryResult, error) { + req := newagentmodel.TaskQueryRequest{ + UserID: userID, + Quadrant: params.Quadrant, + SortBy: params.SortBy, + Order: params.Order, + Limit: params.Limit, + IncludeCompleted: params.IncludeCompleted, + Keyword: params.Keyword, + DeadlineBefore: params.DeadlineBefore, + DeadlineAfter: params.DeadlineAfter, + } + records, err := agentService.QueryTasksForTool(ctx, req) + if err != nil { + return nil, err + } + results := make([]newagentmodel.TaskQueryResult, 0, len(records)) + for _, r := range records { + deadlineStr := "" + if r.DeadlineAt != nil { + deadlineStr = r.DeadlineAt.In(time.Local).Format("2006-01-02 15:04") + } + results = append(results, newagentmodel.TaskQueryResult{ + ID: r.ID, + Title: r.Title, + PriorityGroup: r.PriorityGroup, + IsCompleted: r.IsCompleted, + DeadlineAt: deadlineStr, + }) + } + return results, nil + } +} + +func buildAPIHandlers( + userService *service.UserService, + taskService *service.TaskService, + taskClassService *service.TaskClassService, + courseService *service.CourseService, + scheduleService *service.ScheduleService, + agentService *service.AgentService, + memoryModule *memory.Module, +) *api.ApiHandlers { + return &api.ApiHandlers{ + UserHandler: api.NewUserHandler(userService), + TaskHandler: api.NewTaskHandler(taskService), + TaskClassHandler: api.NewTaskClassHandler(taskClassService), + CourseHandler: api.NewCourseHandler(courseService), + ScheduleHandler: api.NewScheduleAPI(scheduleService), + AgentHandler: api.NewAgentHandler(agentService), + MemoryHandler: api.NewMemoryHandler(memoryModule), + } +} + +func (r *appRuntime) startWorkers(ctx context.Context) { + if r == nil { + return } - r := routers.RegisterRouters(handlers, cacheRepo, userRepo, limiter) - routers.StartEngine(r) + if r.eventBus != nil { + if err := r.registerEventHandlers(); err != nil { + log.Fatalf("Failed to register outbox event handlers: %v", err) + } + r.eventBus.Start(ctx) + log.Println("Outbox event bus started") + } else { + log.Println("Outbox event bus is disabled") + } + + if r.memoryModule != nil { + r.memoryModule.StartWorker(ctx) + } +} + +func (r *appRuntime) registerEventHandlers() error { + // 调用目的：worker/all 启动时复用同一套核心事件注册顺序，避免未来新增入口后复制多份 handler 接线。 + return eventsvc.RegisterCoreOutboxHandlers(r.eventBus, r.outboxRepo, r.manager, r.agentRepo, r.cacheRepo, r.memoryModule) +} + +func (r *appRuntime) startHTTP() { + router := routers.RegisterRouters(r.handlers, r.cacheRepo, r.userRepo, r.limiter) + routers.StartEngine(router) +} + +func (r *appRuntime) close() { + if r == nil { + return + } + if r.eventBus != nil { + r.eventBus.Close() + } } func stringPtrOrNil(value string) *string { diff --git a/backend/cmd/worker/main.go b/backend/cmd/worker/main.go new file mode 100644 index 0000000..bd99347 --- /dev/null +++ b/backend/cmd/worker/main.go @@ -0,0 +1,7 @@ +package main + +import "github.com/LoveLosita/smartflow/backend/cmd" + +func main() { + cmd.StartWorker() +} diff --git a/backend/infra/outbox/engine.go b/backend/infra/outbox/engine.go index a053b5d..3099086 100644 --- a/backend/infra/outbox/engine.go +++ b/backend/infra/outbox/engine.go @@ -144,7 +144,33 @@ func (e *Engine) Start(ctx context.Context) { log.Printf("Kafka topic is ready: %s", e.topic) } + e.StartDispatch(ctx) + e.StartConsume(ctx) +} + +// StartDispatch 单独启动 outbox -> Kafka 的投递循环。 +// +// 职责边界： +// 1. 只负责启动 dispatch 后台 goroutine，不负责启动 Kafka 消费； +// 2. 不重复执行 Start 中的 topic readiness 等待，避免改变原 Start(ctx) 的启动语义； +// 3. ctx 取消后由内部循环自行退出，调用方无需额外停止 goroutine。 +func (e *Engine) StartDispatch(ctx context.Context) { + if e == nil { + return + } go e.startDispatchLoop(ctx) +} + +// StartConsume 单独启动 Kafka -> handler 的消费循环。 +// +// 职责边界： +// 1. 只负责启动 consume 后台 goroutine，不负责扫描或投递 outbox； +// 2. 不注册业务 handler，handler 仍由 RegisterEventHandler 显式注入； +// 3. ctx 取消或 consumer 返回 context.Canceled 时，内部循环按既有逻辑退出。 +func (e *Engine) StartConsume(ctx context.Context) { + if e == nil { + return + } go e.startConsumeLoop(ctx) } diff --git a/backend/infra/outbox/event_bus.go b/backend/infra/outbox/event_bus.go index dd53529..423ff4d 100644 --- a/backend/infra/outbox/event_bus.go +++ b/backend/infra/outbox/event_bus.go @@ -78,6 +78,32 @@ func (b *EventBus) Start(ctx context.Context) { b.engine.Start(ctx) } +// StartDispatch 单独启动事件总线的 outbox 投递循环。 +// +// 职责边界： +// 1. 只暴露 relay/dispatch 运行职责，便于独立进程只负责投递； +// 2. 不启动消费循环，避免与独立 consumer 进程争抢职责； +// 3. 不改变 Start(ctx) 的既有组合启动行为。 +func (b *EventBus) StartDispatch(ctx context.Context) { + if b == nil || b.engine == nil { + return + } + b.engine.StartDispatch(ctx) +} + +// StartConsume 单独启动事件总线的 Kafka 消费循环。 +// +// 职责边界： +// 1. 只暴露 consumer 运行职责，便于独立进程只负责消费； +// 2. 不扫描 outbox、不投递 Kafka，状态推进仍复用 Engine 既有逻辑； +// 3. handler 注册仍由调用方在启动前显式完成。 +func (b *EventBus) StartConsume(ctx context.Context) { + if b == nil || b.engine == nil { + return + } + b.engine.StartConsume(ctx) +} + // Close 关闭事件总线资源（producer/consumer）。 func (b *EventBus) Close() { if b == nil || b.engine == nil { diff --git a/backend/service/events/core_outbox_handlers.go b/backend/service/events/core_outbox_handlers.go new file mode 100644 index 0000000..179b325 --- /dev/null +++ b/backend/service/events/core_outbox_handlers.go @@ -0,0 +1,87 @@ +package events + +import ( + "errors" + "fmt" + + "github.com/LoveLosita/smartflow/backend/dao" + outboxinfra "github.com/LoveLosita/smartflow/backend/infra/outbox" + "github.com/LoveLosita/smartflow/backend/memory" +) + +// RegisterCoreOutboxHandlers 注册核心业务 outbox handler。 +// +// 职责边界： +// 1. 只负责聚合注册当前核心业务 handler，便于 start / worker/all 等启动入口复用同一套接线顺序。 +// 2. 不负责创建 eventBus/outboxRepo/DAO/memoryModule，也不负责启动或关闭事件总线。 +// 3. 不改变单个 Register* 函数的职责；具体 payload 解析、幂等消费和业务落库仍由各自 handler 负责。 +// 4. 入口先完整校验依赖，避免注册到一半才发现依赖缺失，导致事件总线处于半注册状态。 +func RegisterCoreOutboxHandlers( + eventBus *outboxinfra.EventBus, + outboxRepo *outboxinfra.Repository, + repoManager *dao.RepoManager, + agentRepo *dao.AgentDAO, + cacheRepo *dao.CacheDAO, + memoryModule *memory.Module, +) error { + if err := validateCoreOutboxHandlerDeps(eventBus, outboxRepo, repoManager, agentRepo, cacheRepo, memoryModule); err != nil { + return err + } + + // 1. 按照现有 start.go 的接线顺序注册，保证迁移到 worker/all 后消费行为不发生隐式变化。 + // 2. 每一步只包一层业务语义错误，便于启动日志直接定位是哪类 handler 注册失败。 + if err := RegisterChatHistoryPersistHandler(eventBus, outboxRepo, repoManager); err != nil { + return fmt.Errorf("注册聊天历史持久化 handler 失败: %w", err) + } + if err := RegisterTaskUrgencyPromoteHandler(eventBus, outboxRepo, repoManager); err != nil { + return fmt.Errorf("注册任务紧急度平移 handler 失败: %w", err) + } + if err := RegisterChatTokenUsageAdjustHandler(eventBus, outboxRepo, repoManager); err != nil { + return fmt.Errorf("注册会话 token 调整 handler 失败: %w", err) + } + if err := RegisterAgentStateSnapshotHandler(eventBus, outboxRepo, repoManager); err != nil { + return fmt.Errorf("注册 agent 状态快照 handler 失败: %w", err) + } + if err := RegisterAgentTimelinePersistHandler(eventBus, outboxRepo, agentRepo, cacheRepo); err != nil { + return fmt.Errorf("注册 agent 时间线持久化 handler 失败: %w", err) + } + if err := RegisterMemoryExtractRequestedHandler(eventBus, outboxRepo, memoryModule); err != nil { + return fmt.Errorf("注册记忆抽取 handler 失败: %w", err) + } + + return nil +} + +// validateCoreOutboxHandlerDeps 校验核心 outbox handler 聚合注册所需依赖。 +// +// 职责边界： +// 1. 只做 nil 校验，不做数据库、Redis、Kafka 连通性探测，避免注册函数承担启动健康检查职责。 +// 2. 返回 error 表示依赖缺失；返回 nil 表示可以安全进入逐项注册流程。 +func validateCoreOutboxHandlerDeps( + eventBus *outboxinfra.EventBus, + outboxRepo *outboxinfra.Repository, + repoManager *dao.RepoManager, + agentRepo *dao.AgentDAO, + cacheRepo *dao.CacheDAO, + memoryModule *memory.Module, +) error { + if eventBus == nil { + return errors.New("event bus is nil") + } + if outboxRepo == nil { + return errors.New("outbox repository is nil") + } + if repoManager == nil { + return errors.New("repo manager is nil") + } + if agentRepo == nil { + return errors.New("agent repo is nil") + } + if cacheRepo == nil { + return errors.New("cache repo is nil") + } + if memoryModule == nil { + return errors.New("memory module is nil") + } + return nil +} diff --git a/docs/backend/微服务四步迁移与第二阶段并行开发计划.md b/docs/backend/微服务四步迁移与第二阶段并行开发计划.md new file mode 100644 index 0000000..6f1723a --- /dev/null +++ b/docs/backend/微服务四步迁移与第二阶段并行开发计划.md @@ -0,0 +1,557 @@ +# 微服务四步迁移与第二阶段并行开发计划 + +## 1. 文档目的 + +本文档回答两个问题： + +1. 当前 Gin 单体如何平滑演进到“同仓库、多 Go module、分别启动”的微服务形态。 +2. 第二阶段 `to5.1` 主动调度闭环应在什么时间点介入，才能与底座迁移并行推进，避免互相干扰。 + +本轮迁移的核心原则是：**先拆运行边界，再拆服务边界，最后再拆数据所有权**。 + +也就是说，第一轮不追求一步到位变成完整微服务，而是先把现在所有能力都挤在 `cmd.Start()` 里的问题解决掉，让 API、Worker、通知、主动调度后续可以各走各的启动链路。 + +--- + +## 2. 最终目标形态 + +项目保持一个 GitHub 仓库，但后端逐步演进为多个可独立构建、独立启动、独立部署的 Go module。 + +推荐最终目录形态如下： + +```text +SmartFlow-Agent/ + frontend/ + package.json + src/ + + backend/ + apps/ + api/ # Gin BFF / API Gateway + go.mod + cmd/ + api/ + main.go + internal/ + + services/ + notification/ # go-zero 通知服务，飞书优先落地 + go.mod + cmd/ + internal/ + + active-scheduler/ # 主动调度服务/worker + go.mod + cmd/ + internal/ + + schedule/ # 正式日程应用服务，后期拆 + go.mod + cmd/ + internal/ + + task/ # 四象限任务服务，后期拆 + go.mod + cmd/ + internal/ + + workers/ + outbox-worker/ # outbox relay + Kafka consumer + go.mod + cmd/ + internal/ + + agent-worker/ # 后续 agent 重计算/主动优化 worker + go.mod + cmd/ + internal/ + + shared/ + proto/ # RPC 契约 + events/ # Kafka 事件契约、JSON 示例、版本说明 + packages/ # 极少量稳定公共库 + + deploy/ + docs/ +``` + +但这只是最终形态，不是第一轮要直接改成这样。 + +迁移期建议先保持当前 `backend/go.mod`，只在现有模块内拆出多启动入口。等第一轮稳定后，再把 `notification` 作为第一个独立 Go module 放到 `backend/services/notification`。 + +--- + +## 3. 四步走总览 + +### 第一步：拆运行边界，但保持业务行为不变 + +目标：把现在的 Gin 单体拆成同 Go module 内的多启动入口。 + +建议目录： + +```text +backend/ + go.mod + cmd/ + api/ + main.go + worker/ + main.go + all/ + main.go + internal/app/ # 或 cmd/app，承载启动装配函数 +``` + +启动角色： + +```text +api + 启动 Gin、鉴权、SSE、查询接口、用户确认接口、事件发布能力。 + 迁移第一阶段仍保留 API 所需的同步 service/dao，不是最终纯网关。 + +worker + 只启动 outbox relay、Kafka consumer、事件 handler、memory worker。 + +all + 保持当前单体行为，用于本地开发和迁移期兜底。 +``` + +这一阶段必须保证： + +1. `all` 模式行为与当前 `cmd.Start()` 一致。 +2. `api` 进程保留现有同步业务依赖，但不注册业务消费者，不启动 `memoryModule.StartWorker()`。 +3. `worker` 进程不注册 Gin 路由，不占用 HTTP 端口。 +4. 所有现有 API 路径、响应格式、缓存策略、数据库写入语义不变。 +5. 只改启动装配，不改主动调度业务逻辑。 + +这一阶段完成后，第二阶段新功能就可以优先挂到 worker 或事件链路，而不是继续塞进 Gin 主进程。 + +### 第二步：建立事件契约、主动调度闭环 MVP 与飞书触达 + +目标：让“四象限任务池 + 课表时间轴”进入同一个调度闭环，并通过飞书主动触达用户。 + +关键修正： + +1. 主动调度不是等用户打开聊天后才触发，而是由后台 worker 定时/事件驱动触发。 +2. 飞书不是后置锦上添花，而是本周期主动出击闭环的第一版触达渠道。 +3. 第一版飞书只负责通知用户“系统发现问题并生成了建议”，不承载复杂调度判断，也不直接改正式日程。 + +主动调度闭环 MVP： + +```text +后台监控/事件触发 + -> 读取四象限任务 + -> 读取课表/日程空闲时间 + -> 生成局部调整建议 + -> 附带 Reasoning + -> 写入对比预览 + -> 发布 notification.feishu.requested + -> 飞书提醒用户回到系统确认 + -> 用户按变更项确认 + -> 确认后应用 +``` + +建议新增事件契约： + +```text +active_schedule.triggered +schedule.preview.generated +schedule.apply.requested +schedule.apply.succeeded +schedule.apply.failed +notification.feishu.requested +``` + +这一阶段主动调度可以仍在 `backend` 模块内实现，但要按未来服务边界写： + +1. API 只负责测试触发、查询预览、用户确认和正式应用入口。 +2. Worker 负责后台监控、消费触发事件、生成建议、写预览、发布飞书通知事件。 +3. 正式应用仍先复用当前强一致落库链路，不拆成远程服务。 +4. 飞书第一版可以先在 `backend` worker 内实现 webhook/provider，后续再迁出到独立 notification 服务。 + +### 第三步：拆出第一个独立 Go module：notification + +目标：把第二步里先落在 `backend` worker 内的飞书通知，演进为第一个真正服务化模块。 + +推荐目录： + +```text +backend/ + services/ + notification/ + go.mod + cmd/ + notification/ + main.go + internal/ + consumer/ + domain/ + repo/ + provider/ + feishu/ +``` + +推荐技术选型： + +1. 使用 go-zero 作为服务工程框架。 +2. 通过 Kafka 消费 `notification.feishu.requested`。 +3. 若需要同步管理接口，再补 go-zero API 或 RPC。 + +通知服务必须有自己的投递模型，不能只依赖 outbox 的 `consumed` 状态。 + +建议新增通知记录表或等价存储： + +```text +notification_records + id + user_id + channel + biz_type + biz_id + idempotency_key + status + retry_count + last_error + requested_at + sent_at +``` + +飞书发送链路： + +```text +notification.feishu.requested + -> 写入/读取 notification_records + -> 幂等判断 + -> 调用飞书 provider + -> 记录 sent / failed / retry +``` + +这一阶段完成后，飞书挂了不会影响 Gin API，也不会影响主动调度生成预览。 + +> 说明：第二步允许先做简版飞书 webhook，是为了尽快跑通“后台主动发现 -> 飞书触达 -> 用户回系统确认”的产品闭环。第三步再把通知投递模型、失败补偿、幂等记录独立出来，避免第一轮就被完整 notification 平台拖慢。 + +### 第四步：逐步拆主动调度、日程、任务服务 + +目标：在业务边界稳定后，再拆核心服务。 + +推荐顺序： + +```text +active-scheduler + -> schedule + -> task +``` + +拆 `active-scheduler` 的前提： + +1. 主动调度输入输出 DTO 稳定。 +2. Reasoning 结构稳定。 +3. 预览生成和确认协议稳定。 +4. 调度触发事件稳定。 + +拆 `schedule` 的前提： + +1. 正式日程应用命令契约稳定。 +2. 已明确 `schedule_events`、`schedules`、`task_items.embedded_time` 的一致性边界。 +3. 有 apply id / idempotency key。 +4. 有冲突失败回执和回滚策略。 + +拆 `task` 的前提： + +1. 四象限任务池语义稳定。 +2. DDL、完成状态、优先级平移规则稳定。 +3. 任务表所有权明确。 + +在这些前提未满足之前，不建议把 `schedule` 和 `task` 强行拆成独立服务，否则容易变成分布式单体。 + +--- + +## 4. 第二阶段功能如何并行介入 + +第二阶段 `to5.1` 的业务目标是让四象限任务和课表联通，不再是两个孤岛。 + +建议按以下时间点介入。 + +### 介入点 A：第一步完成后，开始“后台主动调度 + 飞书触达”MVP + +当 `api / worker / all` 三种启动角色跑通后，就可以开始主动调度 MVP。 + +这一版不要只做“用户打开聊天后触发”，而是要把后台触发作为主链路，把 API 测试接口作为调试入口。 + +此时你可以开发： + +1. 主动调度后台触发器。 +2. 主动调度 dry-run / trigger 测试接口。 +3. `mock_now` 时间注入。 +4. 任务未完成 / 用户反馈很累 / DDL 临近 的触发 payload。 +5. 写入对比预览。 +6. 发布 `notification.feishu.requested`。 +7. 简版飞书 webhook/provider，通知用户回系统确认。 + +不建议此时开发： + +1. 飞书内直接确认/改日程。 +2. 飞书作为完整 agent 聊天入口。 +3. DDL 全自动插空的复杂版本。 +4. 大范围全局重排。 +5. 独立 schedule/task 微服务。 + +原因：这一阶段的目标是验证“后台主动发现 -> 生成建议预览 -> 飞书触达 -> 用户回系统确认”的闭环，而不是一次性做完所有渠道与触发场景。 + +### 介入点 B：飞书触达可用后，预埋飞书聊天链路 + +当飞书 webhook 通知可用后，可以提前预埋下周期“飞书接入 agent 聊天链路”的边界。 + +下周期目标类似“飞书内直接和 agent 对话”，但本周期只埋以下能力： + +1. `channel` 字段：区分 `web`、`feishu`、未来移动端等入口。 +2. `external_user_id` / `open_id` 映射：把飞书用户身份映射到系统用户。 +3. `external_conversation_id` 映射：把飞书会话映射到系统 `conversation_id`。 +4. 入站消息 DTO：把飞书消息统一转换成 `AgentInboundMessage`。 +5. 出站消息 DTO：把 agent 回复统一转换成 `AgentOutboundMessage`。 +6. 幂等键：飞书 message id 需要映射为入站消息幂等键，避免重复回调导致重复对话。 + +本周期不做： + +1. 飞书内完整多轮 Agent Chat。 +2. 飞书内复杂确认卡片。 +3. 飞书内直接应用日程。 +4. 移动端替代方案。 + +推荐预埋事件： + +```text +agent.channel.message.received +agent.channel.reply.requested +``` + +飞书主动通知仍然走： + +```text +notification.feishu.requested +``` + +也就是说，**通知通道** 和 **聊天通道** 从事件名和 DTO 上就要分开，避免下周期把飞书消息接入时污染 notification 逻辑。 + +### 介入点 C：主动调度 MVP 稳定后，扩充更多主动触发源 + +此时可以扩展： + +1. DDL 临近插入课表空余时间。 +2. 任务未完成监控。 +3. 用户反馈很累后的局部微调。 +4. 提前完成后的碎片任务建议。 +5. 休息/发呆选项。 + +所有触发源都应该进入同一个主动调度入口： + +```text +active_schedule.triggered +``` + +不要每个触发源各写一套调度逻辑。 + +### 介入点 D：飞书通知模型稳定后，拆 notification 独立 module + +当飞书通知从“能发 webhook”升级为“有幂等、有记录、有失败补偿”后，再把它从 `backend` 中迁到： + +```text +backend/services/notification +``` + +迁移时要保持： + +1. 主动调度 worker 只发布 `notification.feishu.requested`，不直接依赖飞书 SDK 或 webhook 细节。 +2. notification 服务负责通知记录、幂等、重试、provider 调用。 +3. 飞书聊天链路如果下周期启动，应单独走 `agent.channel.*` 事件，不混入 notification 投递模型。 + +### 介入点 E：主动调度协议稳定后，再拆 active-scheduler 独立 module + +当主动调度闭环稳定后，再把它从 `backend` 中迁到： + +```text +backend/services/active-scheduler +``` + +迁移时要保持： + +1. API 仍只发布事件和查询预览。 +2. active-scheduler 只负责建议生成和预览。 +3. 正式日程应用仍通过稳定命令或事件进入 schedule 域。 + +--- + +## 4.1 为下周期飞书 Agent 聊天链路预埋 + +如果后续暂缓移动端开发，把飞书作为轻量移动入口，那么本周期接飞书时要提前避免两个误区： + +1. 不要把飞书通知写成只会发送固定文案的死逻辑。 +2. 不要把飞书聊天直接塞进 notification handler。 + +建议从本周期就区分三层： + +```text +notification + 负责主动通知，例如“我发现你的今晚安排可能过载，已生成一版建议”。 + +channel adapter + 负责不同渠道的入站/出站消息适配，例如 web、feishu、未来移动端。 + +agent conversation + 负责真正的 agent 多轮对话、上下文、工具调用、确认卡片。 +``` + +飞书作为聊天入口时，推荐链路是： + +```text +飞书回调 + -> feishu channel adapter + -> 校验签名/解密/去重 + -> 映射系统 user_id 与 conversation_id + -> 发布 agent.channel.message.received + -> agent worker / agent service 处理 + -> 发布 agent.channel.reply.requested + -> feishu channel adapter 发送回复 +``` + +本周期最小预埋： + +1. 配置层预留 `feishu.enabled`、`feishu.webhook`、`feishu.appID`、`feishu.appSecret` 等字段位置。 +2. DTO 层预留 `Channel`、`ExternalUserID`、`ExternalConversationID`、`MessageID`、`IdempotencyKey`。 +3. 事件层预留 `agent.channel.message.received` 和 `agent.channel.reply.requested`。 +4. 数据层先不强行建全量飞书会话表；若需要，只建轻量映射表或先通过 Redis/配置映射过渡。 + +这样下周期即使做“飞书内直接聊 agent”，也不会推翻本周期的飞书通知实现。 + +--- + +## 5. 并行开发分工建议 + +### Codex 优先负责 + +1. 拆 `cmd.Start()`，抽启动装配层。 +2. 新增 `api / worker / all` 启动入口。 +3. 把 outbox relay、Kafka consumer、memory worker 从 API 入口移走。 +4. 为 notification 服务预留事件契约和目录位置。 +5. 补迁移文档和启动说明。 + +### 业务开发优先负责 + +1. 定义主动调度 MVP 的产品语义。 +2. 明确 Reasoning 展示格式。 +3. 明确用户确认粒度。 +4. 设计后台触发、dry-run、trigger 测试场景。 +5. 梳理哪些任务可被退回任务池，哪些必须保护。 +6. 明确飞书通知文案与“回系统确认”的跳转语义。 + +### 双方交汇点 + +交汇点只放在事件契约和 DTO 上。 + +建议先稳定以下结构： + +```text +ActiveScheduleTrigger +ActiveScheduleSuggestion +ActiveScheduleChangeItem +ActiveScheduleReasoning +SchedulePreviewVersion +NotificationRequested +AgentInboundMessage +AgentOutboundMessage +``` + +只要这些结构稳定，底层是否已经拆成独立 module，不影响你继续开发业务。 + +--- + +## 6. 每一步的验收标准 + +### 第一步验收：启动边界 + +1. `all` 模式与当前单体行为一致。 +2. `api` 模式能启动 Gin，并能访问现有接口。 +3. `worker` 模式能启动 outbox 和 memory worker，不启动 HTTP。 +4. API 进程发布的 outbox 消息能被 worker 消费。 +5. 停掉 worker 时，API 仍可启动，只是异步能力延迟执行。 + +### 第二步验收：主动调度 MVP + +1. worker 能后台触发主动调度。 +2. 测试接口能触发同一条主动调度链路。 +3. 能传入 `mock_now`。 +4. 能读取四象限任务和课表空余时间。 +5. 能生成调整建议。 +6. 每个建议都有 Reasoning。 +7. 结果写入对比预览，不直接落库。 +8. 能发布并发送飞书通知，提醒用户回系统确认。 + +### 第三步验收：飞书通知服务 + +1. `notification.feishu.requested` 能被独立服务消费。 +2. 有通知幂等键。 +3. 有通知发送记录。 +4. 飞书失败可重试。 +5. 飞书服务停止不影响 API 和主动调度预览生成。 + +### 第四步验收：核心服务化 + +1. active-scheduler 可独立启动。 +2. schedule/task 的数据所有权逐步明确。 +3. API 不再直接承担核心调度重计算。 +4. 服务间通信只通过 RPC / Kafka / 明确契约完成。 + +--- + +## 7. 风险与回退策略 + +### 风险 1：启动拆分后本地开发变复杂 + +回退策略： + +1. 保留 `all` 模式。 +2. 本地默认仍可一键启动。 +3. 只有联调 worker / 飞书时才分进程启动。 + +### 风险 2：API 不启动消费者后 outbox 堆积 + +回退策略： + +1. 本地使用 `all` 模式。 +2. 联调环境明确启动 worker。 +3. 增加 outbox 堆积观测或临时查询 SQL。 + +### 风险 3：多个 worker 重复投递 + +回退策略： + +1. 第一阶段只启动一个 worker。 +2. 后续再补 outbox claim / processing 状态。 +3. 外部通知必须有业务幂等键。 + +### 风险 4：过早拆 schedule/task 导致分布式单体 + +回退策略： + +1. 第一阶段不拆正式日程落库。 +2. 第二阶段只拆主动建议生成。 +3. schedule/task 等数据所有权稳定后再拆。 + +--- + +## 8. 推荐近期执行顺序 + +近期建议按以下顺序推进： + +1. Codex 先完成启动边界拆分：`api / worker / all`。 +2. 业务侧同步定义主动调度 MVP 的 DTO 与 Reasoning 格式。 +3. 在 `backend` worker 内实现后台主动触发器，并保留 dry-run / trigger 测试接口。 +4. 主动调度结果先写现有对比预览。 +5. 增加 `notification.feishu.requested` 事件，并实现简版飞书 webhook/provider。 +6. 预留 `agent.channel.*` 事件和 channel DTO，为下周期飞书聊天链路做准备。 +7. 飞书通知模型稳定后，新建 `backend/services/notification`，使用 go-zero 落地独立通知服务。 +8. 主动调度稳定后，再考虑拆 `active-scheduler`。 + +一句话总结： + +> 先让项目从“所有能力绑在一个 Gin 单体进程”变成“API 与 Worker 分开跑”；再让第二阶段主动调度闭环挂到 Worker 与事件契约上；最后把飞书通知作为第一个独立 Go module 服务拆出去。