// Package app 是应用的装配根:把 config、logger、DB 池、各业务模块、HTTP // 路由按依赖顺序拼起来,并暴露 Run 给入口程序。本包只做组合,不引入新业务。 // // 启动顺序: // 1. 用 config.Env 派生 logger(FromEnv 在生产环境默认开启 source)。 // 2. 跑 migrations(先 migrate 再开池,避免应用启动时表还没建好;migrate // 失败直接返错由调用方记录、退出)。 // 3. 创建 *db.Pool(实际是 *pgxpool.Pool 别名),失败时不再继续。 // 4. 模块装配:audit recorder -> user repo+service(注入 audit)-> notify // repo+dispatcher(注入 dummy notifier)-> notify handler(复用 userSvc // 作为 SessionResolver,user.Service 接口本身满足该契约)。 // 5. 路由:httpx.NewRouter 返回 chi.Router 接口,直接挂 RequestID/Recover/ // Logger 三个全局中间件,再让模块 Mount 自己的子路由。 // 6. 可选 bootstrap:仅当配置里同时给了 admin 邮箱/密码时尝试。失败仅 // Warn 不阻塞启动。 // 7. 把所有依赖封装到 *App,由 Run 负责启动 http server 与优雅关停。 package app import ( "context" "embed" "errors" "fmt" "io/fs" "log/slog" "net/http" "os" "path/filepath" "time" "github.com/google/uuid" "github.com/yan1h/agent-coding-workflow/internal/audit" "github.com/yan1h/agent-coding-workflow/internal/chat" "github.com/yan1h/agent-coding-workflow/internal/config" "github.com/yan1h/agent-coding-workflow/internal/infra/crypto" "github.com/yan1h/agent-coding-workflow/internal/infra/db" "github.com/yan1h/agent-coding-workflow/internal/infra/git" "github.com/yan1h/agent-coding-workflow/internal/infra/llm" "github.com/yan1h/agent-coding-workflow/internal/infra/logger" "github.com/yan1h/agent-coding-workflow/internal/infra/notify" "github.com/yan1h/agent-coding-workflow/internal/infra/storage" "github.com/yan1h/agent-coding-workflow/internal/project" httpx "github.com/yan1h/agent-coding-workflow/internal/transport/http" "github.com/yan1h/agent-coding-workflow/internal/transport/http/middleware" "github.com/yan1h/agent-coding-workflow/internal/user" "github.com/yan1h/agent-coding-workflow/internal/workspace" ) // App 持有运行期所有需要被关停的资源。字段保持非导出,避免外部直接操作。 type App struct { cfg *config.Config log *slog.Logger pool *db.Pool server *http.Server dispatcher *notify.Dispatcher } // New 把所有模块按启动顺序拼装好。任何阶段失败都会返回错误,不留半成品资源。 // dist 是 cmd/server 用 //go:embed 注入的前端构建产物,根目录为 web/dist; // fs.Sub 之后交给 SPA handler 提供静态文件 + index.html 回退。 func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) { log := logger.FromEnv(cfg.Env, nil) // 跑 migrations 必须在开池前;否则首次启动连不存在的表会让 ping 失败。 if err := db.Migrate(ctx, cfg.DB.DSN, "file://migrations"); err != nil { return nil, fmt.Errorf("migrate: %w", err) } pool, err := db.NewPool(ctx, cfg.DB.DSN) if err != nil { return nil, err } auditRec := audit.NewPostgresRecorder(pool) userRepo := user.NewPostgresRepository(pool) userSvc := user.NewService(userRepo, auditRec) projectRepo := project.NewPostgresRepository(pool) projectSvc := project.NewProjectService(projectRepo, auditRec) requirementSvc := project.NewRequirementService(projectRepo, auditRec) issueSvc := project.NewIssueService(projectRepo, auditRec) notifyRepo := notify.NewPostgresRepository(pool) notifyDispatcher := notify.NewDispatcher(notifyRepo, log) notifyDispatcher.Register(notify.NewDummyNotifier(log)) spaSub, err := fs.Sub(dist, "web/dist") if err != nil { return nil, fmt.Errorf("sub fs: %w", err) } spaHandler := httpx.NewSPAHandler(spaSub) r := httpx.NewRouter(httpx.RouterDeps{ HealthCheck: func(req *http.Request) error { return db.HealthCheck(req.Context(), pool) }, ReadyCheck: func(req *http.Request) error { return db.HealthCheck(req.Context(), pool) }, SPAHandler: spaHandler, }) // chi.Router 接口本身已经声明 Use,无需类型断言到 *chi.Mux。 r.Use(middleware.RequestID) r.Use(middleware.Recover(log)) r.Use(middleware.Logger(log)) // userSvc 直接满足 middleware.SessionResolver(ResolveSession 同名同签名)。 user.NewHandler(userSvc).Mount(r) notify.NewHandler(notifyRepo, userSvc).Mount(r) project.NewHandler(projectSvc, requirementSvc, issueSvc, userSvc, userAdminAdapter{svc: userSvc}).Mount(r) // ===== workspace 模块装配 ===== // 依赖图: // git.Runner (cfg) ─┐ // crypto.Encryptor ─┼─ workspaceService (clones=nil) ─┐ // workspace.Repo ───┘ ├─ CloneRunner ─┐ // │ │ // SetScheduler ───┴───────────────┘ // │ // WorktreeService / GitOpsService // wsDataDir 是数据根;pathing.MainPath/WorktreePath 内部会再 join "workspaces" 子目录。 // 这里显式 mkdir 出 ${dataDir}/workspaces 让首个 clone 的父目录已经存在。 wsDataDir := cfg.Workspace.DataDir if err := os.MkdirAll(filepath.Join(wsDataDir, "workspaces"), 0o755); err != nil { return nil, fmt.Errorf("mkdir workspace data dir: %w", err) } tmpDir := filepath.Join(wsDataDir, "tmp") // 启动期清掉残留 tmp(可能含上次崩溃留下的 askpass 脚本与 SSH 私钥) _ = os.RemoveAll(tmpDir) if err := os.MkdirAll(tmpDir, 0o700); err != nil { return nil, fmt.Errorf("mkdir workspace tmp: %w", err) } enc, err := crypto.NewEncryptor(cfg.MasterKey) if err != nil { return nil, fmt.Errorf("crypto encryptor: %w", err) } gitr := git.NewDefaultRunner(git.Config{ Binary: cfg.Git.Binary, TmpDir: tmpDir, CloneTimeout: cfg.Workspace.CloneTimeout, FetchTimeout: cfg.Workspace.FetchTimeout, PushTimeout: cfg.Workspace.PushTimeout, OpsTimeout: cfg.Workspace.OpsTimeout, }) wsRepo := workspace.NewPostgresRepository(pool) // 进程启动时把卡死的 cloning/syncing 复位为 error,避免行永远卡住 if err := wsRepo.ResetStuckSyncStatuses(ctx); err != nil { log.Warn("reset stuck workspace sync statuses", "err", err.Error()) } pa := projectAccessAdapter{p: projectSvc} // 先以 nil scheduler 构造 service,cloneRunner 再回填(环依赖:runner 需要凭据访问器,访问器在 service 上) wsSvc := workspace.NewWorkspaceService(wsRepo, auditRec, enc, gitr, pa, nil, wsDataDir) // 共享凭据加载器:CloneRunner 与 GitOpsService 都需要。 // 通过 CredentialAccessor 接口避免直接持有 *workspaceService 私有类型。 credAccessor, ok := wsSvc.(workspace.CredentialAccessor) if !ok { return nil, fmt.Errorf("workspace service does not implement CredentialAccessor") } credLoader := func(ctx context.Context, wsID uuid.UUID) (*git.Credential, error) { return credAccessor.LoadDecryptedCredential(ctx, wsID) } cloneRunner := workspace.NewCloneRunner(wsRepo, auditRec, gitr, log, credLoader, cfg.Workspace.CloneTimeout) if setter, ok := wsSvc.(workspace.SchedulerSetter); ok { setter.SetScheduler(cloneRunner) } wtSvc := workspace.NewWorktreeService(wsRepo, auditRec, gitr, pa, wsDataDir) gopSvc := workspace.NewGitOpsService(wsRepo, auditRec, gitr, pa, credLoader) workspace.NewHandler(wsSvc, wtSvc, gopSvc, userSvc, userAdminAdapter{svc: userSvc}).Mount(r) // ===== chat 模块装配 ===== // 依赖图: // pgx.Pool ─ chat.Repository ─┐ // ├─ EndpointLoader ─ llm.Registry ─┐ // crypto.Encryptor ───────────┘ │ // │ // storage.LocalFS ─ Storage ──┐ │ // ├─ MessageService ─ Handler ──────┤ // notify.Dispatcher ─ StreamerHub ─ ConversationService ─ Hub ───┘ // // 启动期把卡死的 pending 消息复位为 error,避免行永远卡住。 chatRepo := chat.NewRepository(pool) if err := chatRepo.MarkPendingAsErrorOnStartup(ctx); err != nil { log.Warn("mark chat pending messages as error on startup", "err", err.Error()) } endpointLoader := chat.NewEndpointLoader(chatRepo, enc) llmRegistry := llm.NewRegistry(endpointLoader) var attachStorage storage.Storage switch cfg.Storage.Driver { case "localfs", "": s, err := storage.NewLocalFS(cfg.Storage.LocalFS.BasePath) if err != nil { return nil, fmt.Errorf("storage localfs: %w", err) } attachStorage = s default: return nil, fmt.Errorf("unsupported storage driver %q", cfg.Storage.Driver) } // safety_margin_pct 配置为百分比(5.0 表示 5%),构造器期望小数(0.05),故 /100。 safetyFraction := cfg.Chat.History.SafetyMarginPct / 100 retention := time.Duration(cfg.Chat.Stream.RingBufferRetentionSeconds) * time.Second chatEpSvc := chat.NewEndpointService(chatRepo, enc, llmRegistry, auditRec, log) chatTplSvc := chat.NewTemplateService(chatRepo, auditRec, log) chatConvSvc := chat.NewConversationService(chatRepo, llmRegistry, auditRec, log) chatHub := chat.NewStreamerHub(chatRepo, llmRegistry, notifyDispatcher, auditRec, chatConvSvc, log, retention, safetyFraction) chatMsgSvc := chat.NewMessageService(chatRepo, attachStorage, chatHub, auditRec, log, chat.MessageServiceConfig{ MimeWhitelist: cfg.Chat.Attachment.MimeWhitelist, MaxFileBytes: cfg.Chat.Attachment.MaxFileSizeMB << 20, MaxMsgBytes: cfg.Chat.Attachment.MaxMessageSizeMB << 20, SafetyPct: safetyFraction, }) chatUsageSvc := chat.NewUsageService(chatRepo) chatUploader := chat.NewUploader(chatRepo, attachStorage, cfg.Chat.Attachment.MimeWhitelist, cfg.Chat.Attachment.MaxFileSizeMB<<20, ) chat.NewHandler(chatConvSvc, chatMsgSvc, chatTplSvc, chatEpSvc, chatUsageSvc, chatUploader, userSvc, userAdminAdapter{svc: userSvc}).Mount(r) if cfg.Bootstrap.AdminEmail != "" && cfg.Bootstrap.AdminPassword != "" { if u, err := userSvc.Bootstrap(ctx, cfg.Bootstrap.AdminEmail, cfg.Bootstrap.AdminPassword); err != nil { log.Warn("bootstrap admin failed", "err", err.Error()) } else if u != nil { log.Info("bootstrap admin created", "email", u.Email) } } srv := &http.Server{ Addr: cfg.HTTP.Addr, Handler: r, ReadHeaderTimeout: 10 * time.Second, ReadTimeout: 30 * time.Second, WriteTimeout: 60 * time.Second, IdleTimeout: 120 * time.Second, } return &App{cfg: cfg, log: log, pool: pool, server: srv, dispatcher: notifyDispatcher}, nil } // Run 启动 HTTP server 并阻塞直到 ctx 取消(Shutdown)或服务器报错。 // 关停路径上保证关闭 db pool;返回 nil 表示由 ctx 取消触发的正常关停。 func (a *App) Run(ctx context.Context) error { a.log.Info("server starting", "addr", a.cfg.HTTP.Addr, "env", a.cfg.Env) errCh := make(chan error, 1) go func() { if err := a.server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { errCh <- err } }() select { case <-ctx.Done(): a.log.Info("shutdown signal received, draining...") shutCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() shutErr := a.server.Shutdown(shutCtx) if shutErr != nil { a.log.Error("server shutdown", "err", shutErr.Error()) } // server 已停止接收新请求;等异步通知投递结束再关 pool if err := a.dispatcher.Shutdown(shutCtx); err != nil { a.log.Warn("notify dispatcher drain", "err", err.Error()) } a.pool.Close() return shutErr case err := <-errCh: drainCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() _ = a.dispatcher.Shutdown(drainCtx) a.pool.Close() return err } } // userAdminAdapter 把 user.Service 适配为 project.AdminLookup(窄接口)。 // 复用 user.Service.Get:返回的 *User 直接读 IsAdmin。 // workspace.AdminLookup 与 project.AdminLookup 同形(IsAdmin 同名同签名), // 因此本类型同时满足两者,可在两个 handler 复用。 type userAdminAdapter struct{ svc user.Service } func (a userAdminAdapter) IsAdmin(ctx context.Context, id uuid.UUID) (bool, error) { u, err := a.svc.Get(ctx, id) if err != nil { return false, err } return u.IsAdmin, nil } // projectAccessAdapter 把 project.ProjectService.Get / GetByID 桥接成 workspace.ProjectAccess。 // - Resolve = by slug:用于 POST /projects/{slug}/workspaces 等路径 // - ResolveByID = by uuid:用于 wsID 拿到 ws 后再校验 project 权限 // // ACL 语义复刻 project 包内部 assertReadable / assertWritable: // // read private: owner + admin // read internal: 任意已登录 // write *: owner + admin // // 注意:project.assertReadable 在 visibility=private 且非 owner+admin 时返回 NotFound // 防探测;这里 ProjectService.Get/GetByID 已替我们兜住,到本适配层时 pj 一定可读。 type projectAccessAdapter struct { p project.ProjectService } func (a projectAccessAdapter) Resolve(ctx context.Context, callerID uuid.UUID, isAdmin bool, projectSlug string) (uuid.UUID, bool, bool, error) { pj, err := a.p.Get(ctx, project.Caller{UserID: callerID, IsAdmin: isAdmin}, projectSlug) if err != nil { return uuid.Nil, false, false, err } canRead, canWrite := projectACL(pj, callerID, isAdmin) return pj.ID, canRead, canWrite, nil } func (a projectAccessAdapter) ResolveByID(ctx context.Context, callerID uuid.UUID, isAdmin bool, projectID uuid.UUID) (bool, bool, error) { pj, err := a.p.GetByID(ctx, project.Caller{UserID: callerID, IsAdmin: isAdmin}, projectID) if err != nil { return false, false, err } canRead, canWrite := projectACL(pj, callerID, isAdmin) return canRead, canWrite, nil } // projectACL 复制 project 包内部 assertReadable/assertWritable 的语义: // - admin 永远可读可写 // - owner 可读可写 // - internal 可见性:所有登录用户可读 // // 注:写操作还需考虑 archived 状态;workspace 的写场景会在 service 层用 ProjectAccess // 拿到 canWrite 后再下沉,archived 检查由 project service 在 GetByID 已可见。 func projectACL(pj *project.Project, callerID uuid.UUID, isAdmin bool) (canRead, canWrite bool) { owned := pj.OwnerID == callerID canWrite = isAdmin || owned canRead = canWrite || pj.Visibility == project.VisibilityInternal return canRead, canWrite }