You've already forked agentic-coding-workflow
848 lines
33 KiB
Go
848 lines
33 KiB
Go
// 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/acp"
|
|
"github.com/yan1h/agent-coding-workflow/internal/audit"
|
|
"github.com/yan1h/agent-coding-workflow/internal/chat"
|
|
mcp "github.com/yan1h/agent-coding-workflow/internal/mcp"
|
|
"github.com/yan1h/agent-coding-workflow/internal/config"
|
|
"github.com/yan1h/agent-coding-workflow/internal/infra/clock"
|
|
"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/jobs"
|
|
"github.com/yan1h/agent-coding-workflow/internal/jobs/handlers"
|
|
"github.com/yan1h/agent-coding-workflow/internal/jobs/runners"
|
|
"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
|
|
jobs *jobs.Module
|
|
jobsCancel context.CancelFunc
|
|
acpSup *acp.Supervisor // ACP supervisor — ShutdownAll on Run shutdown
|
|
}
|
|
|
|
// 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)
|
|
wrappedAudit := mcp.AuditWrap(auditRec) // 给业务 service 用，自动注入 via_mcp metadata
|
|
|
|
userRepo := user.NewPostgresRepository(pool)
|
|
userSvc := user.NewService(userRepo, auditRec)
|
|
|
|
projectRepo := project.NewPostgresRepository(pool)
|
|
projectSvc := project.NewProjectService(projectRepo, wrappedAudit)
|
|
requirementSvc := project.NewRequirementService(projectRepo, wrappedAudit)
|
|
issueSvc := project.NewIssueService(projectRepo, wrappedAudit)
|
|
|
|
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，避免行永远卡住；
|
|
// 每个被复位的工作区写一条 workspace.sync_aborted_on_restart audit。
|
|
if abortedIDs, err := wsRepo.ResetStuckSyncStatuses(ctx); err != nil {
|
|
log.Warn("reset stuck workspace sync statuses", "err", err.Error())
|
|
} else {
|
|
for _, wsID := range abortedIDs {
|
|
if rerr := auditRec.Record(ctx, audit.Entry{
|
|
Action: "workspace.sync_aborted_on_restart",
|
|
TargetType: "workspace",
|
|
TargetID: wsID.String(),
|
|
}); rerr != nil {
|
|
log.Warn("workspace.sync_aborted_on_restart audit failed",
|
|
"workspace_id", wsID, "err", rerr.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)
|
|
|
|
// ===== MCP 模块装配 =====
|
|
mcpRepo := mcp.NewPostgresRepository(pool)
|
|
mcpTokenSvc := mcp.NewTokenService(mcpRepo, auditRec, nil) // mcp 自身 audit 不需要 via_mcp 标记
|
|
mcpRateLimiter := mcp.NewRateLimiter(mcp.RateLimitConfig{
|
|
PerTokenPerMinute: cfg.MCP.RateLimit.PerTokenPerMinute,
|
|
GlobalPerMinute: cfg.MCP.RateLimit.GlobalPerMinute,
|
|
}, nil)
|
|
mcpDeps := mcp.ServerDeps{
|
|
Caller: mcp.NewCallerResolver(userSvc),
|
|
Projects: projectSvc,
|
|
Reqs: requirementSvc,
|
|
Issues: issueSvc,
|
|
Workspaces: wsSvc,
|
|
Templates: chatTplSvc,
|
|
Messages: chatMsgSvc,
|
|
Conversations: chatConvSvc,
|
|
}
|
|
mcpServer := mcp.NewMCPServer(mcpDeps)
|
|
mcp.MountTransports(r, mcpServer, mcpTokenSvc, mcpRateLimiter)
|
|
mcp.MountAdmin(r, mcp.AdminHandlerDeps{
|
|
Tokens: mcpTokenSvc,
|
|
Users: userSvc,
|
|
})
|
|
|
|
// ===== ACP module wiring (Part 2: full SessionService + Supervisor) =====
|
|
acpRepo := acp.NewPostgresRepository(pool)
|
|
akSvc := acp.NewAgentKindService(acpRepo, enc, auditRec)
|
|
|
|
// Phase 1: startup reaper — recover sessions interrupted by previous crash.
|
|
// ResetStuckSessionsOnRestart marks all 'starting'/'running' sessions as 'crashed'
|
|
// and returns them; we release their worktrees and notify owners.
|
|
abortedSessions, rerr := acpRepo.ResetStuckSessionsOnRestart(ctx)
|
|
if rerr != nil {
|
|
log.Warn("acp.startup_reaper.reset_failed", "err", rerr.Error())
|
|
} else {
|
|
for _, sess := range abortedSessions {
|
|
if sess.IsMainWorktree {
|
|
if _, rerr := acpRepo.ReleaseMainWorktree(ctx, sess.WorkspaceID, sess.ID); rerr != nil {
|
|
log.Warn("acp.startup_reaper.release_main_failed",
|
|
"session_id", sess.ID, "err", rerr.Error())
|
|
}
|
|
} else {
|
|
if _, rerr := wtSvc.ReleaseByHolder(ctx, "session:"+sess.ID.String()); rerr != nil {
|
|
log.Warn("acp.startup_reaper.release_subwt_failed",
|
|
"session_id", sess.ID, "err", rerr.Error())
|
|
}
|
|
}
|
|
uid := sess.UserID
|
|
if rerr := auditRec.Record(ctx, audit.Entry{
|
|
UserID: &uid, Action: "acp.session.aborted_on_restart",
|
|
TargetType: "acp_session", TargetID: sess.ID.String(),
|
|
Metadata: map[string]any{
|
|
"agent_kind_id": sess.AgentKindID.String(),
|
|
"branch": sess.Branch,
|
|
},
|
|
}); rerr != nil {
|
|
log.Warn("acp.startup_reaper.audit_failed",
|
|
"session_id", sess.ID, "err", rerr.Error())
|
|
}
|
|
if derr := notifyDispatcher.Dispatch(ctx, notify.Message{
|
|
UserID: sess.UserID,
|
|
Topic: "acp.session_crashed",
|
|
Severity: notify.SeverityWarning,
|
|
Title: "ACP session terminated by server restart",
|
|
Body: "Your session was interrupted while the server restarted.",
|
|
Metadata: map[string]any{
|
|
"session_id": sess.ID.String(),
|
|
"branch": sess.Branch,
|
|
},
|
|
}); derr != nil {
|
|
log.Warn("acp.startup_reaper.notify_failed",
|
|
"session_id", sess.ID, "err", derr.Error())
|
|
}
|
|
}
|
|
}
|
|
|
|
// Phase 2: build supervisor + session service.
|
|
acpProjAccess := acpProjectAccessAdapter{wsRepo: wsRepo, projectRepo: projectRepo}
|
|
acpSup := acp.NewSupervisor(
|
|
acpRepo, auditRec, notifyDispatcher,
|
|
wtSvc, wsRepo, enc,
|
|
acp.SupervisorConfig{
|
|
SpawnTimeout: cfg.Acp.SpawnTimeout,
|
|
KillGrace: cfg.Acp.KillGrace,
|
|
StderrBufferLines: cfg.Acp.StderrBufferLines,
|
|
StderrTailBytes: cfg.Acp.StderrTailBytes,
|
|
EventMaxPayload: cfg.Acp.EventMaxPayload,
|
|
EventTruncateField: cfg.Acp.EventTruncateField,
|
|
ShutdownGrace: cfg.Acp.ShutdownGrace,
|
|
},
|
|
log,
|
|
)
|
|
sessSvc := acp.NewSessionService(
|
|
acpRepo, wsSvc, wtSvc, wsRepo,
|
|
acpProjAccess, acpSup, auditRec,
|
|
acp.SessionServiceConfig{
|
|
MaxActiveGlobal: cfg.Acp.MaxActiveGlobal,
|
|
MaxActivePerUser: cfg.Acp.MaxActivePerUser,
|
|
},
|
|
)
|
|
acp.NewHandler(akSvc, sessSvc, acpSup, acpRepo, userSvc, userAdminAdapter{svc: userSvc}, enc, acp.WSConfig{
|
|
PingInterval: cfg.Acp.WSPingInterval,
|
|
PongTimeout: cfg.Acp.WSPongTimeout,
|
|
SendBuffer: cfg.Acp.WSSendBuffer,
|
|
}).Mount(r)
|
|
|
|
// ===== jobs module wiring =====
|
|
jobsRepo := jobs.NewPostgresRepository(pool)
|
|
|
|
webhookBackoff := []time.Duration{
|
|
30 * time.Second, 2 * time.Minute, 8 * time.Minute, 30 * time.Minute, 2 * time.Hour,
|
|
}
|
|
|
|
webhookHandler := handlers.NewWebhookHandler(handlers.WebhookConfig{
|
|
URL: cfg.Notify.Webhook.URL,
|
|
Secret: cfg.Notify.Webhook.Secret,
|
|
Timeout: cfg.Notify.Webhook.Timeout,
|
|
}, clock.Real())
|
|
|
|
webhookNotifier := jobs.NewWebhookNotifier(jobsRepo, jobs.WebhookNotifierConfig{
|
|
Enabled: cfg.Notify.Webhook.Enabled,
|
|
Topics: cfg.Notify.Webhook.Topics,
|
|
Backoff: webhookBackoff,
|
|
}, clock.Real(), auditRec, log)
|
|
notifyDispatcher.Register(webhookNotifier)
|
|
|
|
dispAdapter := dispatcherAdapter{d: notifyDispatcher}
|
|
wsFetchRunner := runners.NewWorkspaceFetch(
|
|
workspaceFetchAdapter{repo: wsRepo},
|
|
workspaceFetcherAdapter{gitr: gitr, credLoader: credLoader},
|
|
dispAdapter, log)
|
|
wtPruneRunner := runners.NewWorktreePrune(
|
|
worktreePruneAdapter{wsRepo: wsRepo, pLookup: projectRepo, log: log},
|
|
gitr,
|
|
dispAdapter,
|
|
runners.WorktreePruneConfig{
|
|
IdleThreshold: cfg.Jobs.WorktreePrune.IdleThreshold,
|
|
WarningLead: cfg.Jobs.WorktreePrune.WarningLead,
|
|
}, log, auditRec)
|
|
attCleanupRunner := runners.NewAttachmentCleanup(
|
|
chatAttachmentAdapter{repo: chatRepo},
|
|
attachStorage,
|
|
runners.AttachmentCleanupConfig{Retention: cfg.Jobs.AttachmentCleanup.Retention},
|
|
log)
|
|
jobReaperRunner := runners.NewJobReaper(jobsRepo, cfg.Jobs.JobReaper.LeaseTimeout, log)
|
|
jobPurgeRunner := runners.NewJobPurge(jobsRepo, cfg.Jobs.JobPurge.CompletedRetention, log)
|
|
acpEventsPurgeRunner := runners.NewAcpEventsPurge(
|
|
acpEventsPurgeAdapter{repo: acpRepo}, cfg.Jobs.AcpEventsPurge.Retention, log)
|
|
|
|
deadLetterFn := func(ctx context.Context, job jobs.Job, herr error) {
|
|
errMsg := herr.Error()
|
|
if len(errMsg) > 500 {
|
|
errMsg = errMsg[:500]
|
|
}
|
|
if rerr := auditRec.Record(ctx, audit.Entry{
|
|
Action: "job.dead_letter",
|
|
TargetType: "job",
|
|
TargetID: job.ID.String(),
|
|
Metadata: map[string]any{
|
|
"type": string(job.Type),
|
|
"attempts": job.Attempts,
|
|
"err": errMsg,
|
|
},
|
|
}); rerr != nil {
|
|
log.Warn("dead_letter.audit_failed", "err", rerr.Error())
|
|
}
|
|
admins, err := userSvc.ListAdmins(ctx)
|
|
if err != nil {
|
|
log.Error("dead_letter.list_admins_failed", "err", err.Error())
|
|
return
|
|
}
|
|
for _, a := range admins {
|
|
if derr := notifyDispatcher.Dispatch(ctx, notify.Message{
|
|
UserID: a.ID,
|
|
Topic: "webhook.dead_letter",
|
|
Severity: notify.SeverityError,
|
|
Title: "Webhook delivery failed permanently",
|
|
Body: errMsg,
|
|
Metadata: map[string]any{
|
|
"job_id": job.ID.String(),
|
|
"job_type": string(job.Type),
|
|
},
|
|
CreatedAt: clock.Real().Now().UTC(),
|
|
}); derr != nil {
|
|
log.Warn("dead_letter.dispatch_failed", "admin_id", a.ID, "err", derr.Error())
|
|
}
|
|
}
|
|
}
|
|
|
|
jobsModule := jobs.NewModule(jobs.Config{
|
|
Enabled: cfg.Jobs.Enabled,
|
|
Workers: cfg.Jobs.Workers,
|
|
ShutdownGrace: cfg.Jobs.ShutdownGrace,
|
|
WorkspaceFetch: jobs.RunnerConfig{
|
|
Enabled: cfg.Jobs.WorkspaceFetch.Enabled,
|
|
Interval: cfg.Jobs.WorkspaceFetch.Interval,
|
|
},
|
|
WorktreePrune: jobs.WorktreePruneConfig{
|
|
Enabled: cfg.Jobs.WorktreePrune.Enabled,
|
|
Interval: cfg.Jobs.WorktreePrune.Interval,
|
|
IdleThreshold: cfg.Jobs.WorktreePrune.IdleThreshold,
|
|
WarningLead: cfg.Jobs.WorktreePrune.WarningLead,
|
|
},
|
|
AttachmentCleanup: jobs.AttachmentCleanupConfig{
|
|
Enabled: cfg.Jobs.AttachmentCleanup.Enabled,
|
|
Interval: cfg.Jobs.AttachmentCleanup.Interval,
|
|
Retention: cfg.Jobs.AttachmentCleanup.Retention,
|
|
},
|
|
JobReaper: jobs.JobReaperConfig{
|
|
Enabled: cfg.Jobs.JobReaper.Enabled,
|
|
Interval: cfg.Jobs.JobReaper.Interval,
|
|
LeaseTimeout: cfg.Jobs.JobReaper.LeaseTimeout,
|
|
},
|
|
JobPurge: jobs.JobPurgeConfig{
|
|
Enabled: cfg.Jobs.JobPurge.Enabled,
|
|
Interval: cfg.Jobs.JobPurge.Interval,
|
|
CompletedRetention: cfg.Jobs.JobPurge.CompletedRetention,
|
|
},
|
|
AcpEventsPurge: jobs.AcpEventsPurgeConfig{
|
|
Enabled: cfg.Jobs.AcpEventsPurge.Enabled,
|
|
Interval: cfg.Jobs.AcpEventsPurge.Interval,
|
|
Retention: cfg.Jobs.AcpEventsPurge.Retention,
|
|
},
|
|
}, jobs.ModuleDeps{
|
|
Repo: jobsRepo,
|
|
Handlers: map[jobs.JobType]jobs.Handler{
|
|
jobs.TypeWebhookDeliver: webhookHandler,
|
|
},
|
|
RunnerFns: map[string]jobs.RunnerFunc{
|
|
"workspace_fetch": wsFetchRunner.Run,
|
|
"worktree_prune": wtPruneRunner.Run,
|
|
"attachment_cleanup": attCleanupRunner.Run,
|
|
"job_reaper": jobReaperRunner.Run,
|
|
"job_purge": jobPurgeRunner.Run,
|
|
"acp_events_purge": acpEventsPurgeRunner.Run,
|
|
},
|
|
Clock: clock.Real(),
|
|
Log: log,
|
|
WorkerBackoff: webhookBackoff,
|
|
OnDeadLetter: deadLetterFn,
|
|
})
|
|
|
|
// Start the module with a background context; cancellation is wired into
|
|
// App.Run's shutdown path via jobsCancel.
|
|
jobsCtx, jobsCancel := context.WithCancel(context.Background())
|
|
jobsModule.Start(jobsCtx)
|
|
|
|
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,
|
|
jobs: jobsModule,
|
|
jobsCancel: jobsCancel,
|
|
acpSup: acpSup,
|
|
}, 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())
|
|
}
|
|
// Stop ACP subprocesses (kill child agents) BEFORE jobs/dispatcher drain so
|
|
// any final crash notifications dispatched by onExit can flow through.
|
|
if a.acpSup != nil {
|
|
a.acpSup.ShutdownAll(shutCtx)
|
|
}
|
|
// Stop jobs module first so no in-flight worker can dispatch new
|
|
// notifications after we begin to drain the dispatcher.
|
|
if a.jobs != nil && a.cfg.Jobs.Enabled {
|
|
a.jobsCancel()
|
|
stopCtx, stopCancel := context.WithTimeout(context.Background(), a.cfg.Jobs.ShutdownGrace+5*time.Second)
|
|
a.jobs.Stop(stopCtx)
|
|
stopCancel()
|
|
}
|
|
// server 已停止接收新请求；等异步通知投递结束再关 pool。
|
|
// 用独立的 drainCtx：shutCtx 的 30s 预算已被 server.Shutdown 与
|
|
// jobs.Stop 的实际墙钟时间消耗，复用可能让 dispatcher 拿到已死 ctx
|
|
// 而立即返回，丢失在途的 webhook enqueue（DB 写）。
|
|
drainCtx, drainCancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer drainCancel()
|
|
if err := a.dispatcher.Shutdown(drainCtx); 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()
|
|
if a.acpSup != nil {
|
|
a.acpSup.ShutdownAll(drainCtx)
|
|
}
|
|
if a.jobs != nil && a.cfg.Jobs.Enabled {
|
|
a.jobsCancel()
|
|
stopCtx, stopCancel := context.WithTimeout(context.Background(), a.cfg.Jobs.ShutdownGrace+5*time.Second)
|
|
a.jobs.Stop(stopCtx)
|
|
stopCancel()
|
|
}
|
|
_ = 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
|
|
}
|
|
|
|
// acpProjectAccessAdapter satisfies acp.ProjectAccess. It uses workspace.Repository
|
|
// to resolve workspace → project, then project.Repository for issue/requirement.
|
|
type acpProjectAccessAdapter struct {
|
|
wsRepo workspace.Repository
|
|
projectRepo project.Repository
|
|
}
|
|
|
|
func (a acpProjectAccessAdapter) GetProjectByWorkspace(ctx context.Context, wsID uuid.UUID) (*project.Project, error) {
|
|
ws, err := a.wsRepo.GetWorkspaceByID(ctx, wsID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return a.projectRepo.GetProjectByID(ctx, ws.ProjectID)
|
|
}
|
|
|
|
func (a acpProjectAccessAdapter) GetIssueByNumber(ctx context.Context, projectID uuid.UUID, number int) (*project.Issue, error) {
|
|
return a.projectRepo.GetIssueByNumber(ctx, projectID, number)
|
|
}
|
|
|
|
func (a acpProjectAccessAdapter) GetRequirementByNumber(ctx context.Context, projectID uuid.UUID, number int) (*project.Requirement, error) {
|
|
return a.projectRepo.GetRequirementByNumber(ctx, projectID, number)
|
|
}
|
|
|
|
// ===== Adapters: bridge cross-module types into the narrow interfaces the
|
|
// jobs/runners package expects, so the runners package never imports
|
|
// workspace/chat/notify.
|
|
|
|
// workspaceFetchAdapter implements runners.WorkspaceFetchRepo backed by
|
|
// workspace.Repository. Translates workspace.FetchTarget into
|
|
// runners.WorkspaceFetchTarget (same shape).
|
|
type workspaceFetchAdapter struct{ repo workspace.Repository }
|
|
|
|
func (a workspaceFetchAdapter) ListFetchTargets(ctx context.Context) ([]runners.WorkspaceFetchTarget, error) {
|
|
targets, err := a.repo.ListFetchTargets(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make([]runners.WorkspaceFetchTarget, 0, len(targets))
|
|
for _, t := range targets {
|
|
out = append(out, runners.WorkspaceFetchTarget{
|
|
ID: t.ID, OwnerID: t.OwnerID, Name: t.Name, MainPath: t.MainPath,
|
|
})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (a workspaceFetchAdapter) MarkSyncing(ctx context.Context, id uuid.UUID) (bool, error) {
|
|
return a.repo.MarkSyncing(ctx, id)
|
|
}
|
|
|
|
func (a workspaceFetchAdapter) MarkFetchResult(ctx context.Context, id uuid.UUID, success bool, errMsg string) error {
|
|
return a.repo.MarkFetchResult(ctx, id, success, errMsg)
|
|
}
|
|
|
|
// workspaceFetcherAdapter implements runners.WorkspaceFetcher by loading the
|
|
// workspace credential and calling git.Runner.Fetch directly.
|
|
type workspaceFetcherAdapter struct {
|
|
gitr git.Runner
|
|
credLoader func(ctx context.Context, wsID uuid.UUID) (*git.Credential, error)
|
|
}
|
|
|
|
func (a workspaceFetcherAdapter) Fetch(ctx context.Context, mainPath string, wsID uuid.UUID) error {
|
|
cred, err := a.credLoader(ctx, wsID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return a.gitr.Fetch(ctx, mainPath, cred)
|
|
}
|
|
|
|
// projectLookup is the minimal interface needed by worktreePruneAdapter to
|
|
// look up a project by ID without ACL enforcement (background system job path).
|
|
type projectLookup interface {
|
|
GetProjectByID(ctx context.Context, id uuid.UUID) (*project.Project, error)
|
|
}
|
|
|
|
// worktreePruneAdapter implements runners.WorktreePruneRepo by composing the
|
|
// workspace.Repository's worktree-prune methods with per-worktree workspace+
|
|
// project lookups (to populate OwnerID and MainPath that aren't on the
|
|
// workspace_worktrees row). N+1 acceptable: prune lists are small per tick.
|
|
type worktreePruneAdapter struct {
|
|
wsRepo workspace.Repository
|
|
pLookup projectLookup
|
|
log *slog.Logger
|
|
}
|
|
|
|
func (a worktreePruneAdapter) ListWorktreesNeedingPruneWarning(ctx context.Context, lastUsedBefore time.Time) ([]runners.WorktreeRow, error) {
|
|
rows, err := a.wsRepo.ListWorktreesNeedingPruneWarning(ctx, lastUsedBefore)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return a.enrichWorktrees(ctx, rows), nil
|
|
}
|
|
|
|
func (a worktreePruneAdapter) ListWorktreesNeedingPrune(ctx context.Context, lastUsedBefore time.Time) ([]runners.WorktreeRow, error) {
|
|
rows, err := a.wsRepo.ListWorktreesNeedingPrune(ctx, lastUsedBefore)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return a.enrichWorktrees(ctx, rows), nil
|
|
}
|
|
|
|
func (a worktreePruneAdapter) MarkPruneWarning(ctx context.Context, id uuid.UUID) error {
|
|
return a.wsRepo.MarkPruneWarning(ctx, id)
|
|
}
|
|
|
|
func (a worktreePruneAdapter) TryStartPrune(ctx context.Context, id uuid.UUID) (runners.WorktreeRow, error) {
|
|
wt, err := a.wsRepo.TryStartPrune(ctx, id)
|
|
if err != nil {
|
|
return runners.WorktreeRow{}, err
|
|
}
|
|
return a.enrichOne(ctx, wt), nil
|
|
}
|
|
|
|
func (a worktreePruneAdapter) FinishPruneSuccess(ctx context.Context, id uuid.UUID) error {
|
|
return a.wsRepo.FinishPruneSuccess(ctx, id)
|
|
}
|
|
|
|
func (a worktreePruneAdapter) FinishPruneRollback(ctx context.Context, id uuid.UUID) error {
|
|
return a.wsRepo.FinishPruneRollback(ctx, id)
|
|
}
|
|
|
|
// enrichWorktrees converts []*workspace.Worktree to []runners.WorktreeRow.
|
|
func (a worktreePruneAdapter) enrichWorktrees(ctx context.Context, rows []*workspace.Worktree) []runners.WorktreeRow {
|
|
out := make([]runners.WorktreeRow, 0, len(rows))
|
|
for _, w := range rows {
|
|
out = append(out, a.enrichOne(ctx, w))
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (a worktreePruneAdapter) enrichOne(ctx context.Context, w *workspace.Worktree) runners.WorktreeRow {
|
|
row := runners.WorktreeRow{ID: w.ID, WorkspaceID: w.WorkspaceID, Branch: w.Branch, Path: w.Path}
|
|
ws, err := a.wsRepo.GetWorkspaceByID(ctx, w.WorkspaceID)
|
|
if err != nil {
|
|
a.log.Warn("worktree_prune.adapter.lookup_workspace_failed",
|
|
"worktree_id", w.ID, "workspace_id", w.WorkspaceID, "err", err.Error())
|
|
return row
|
|
}
|
|
row.MainPath = ws.MainPath
|
|
pj, err := a.pLookup.GetProjectByID(ctx, ws.ProjectID)
|
|
if err != nil {
|
|
a.log.Warn("worktree_prune.adapter.lookup_project_failed",
|
|
"worktree_id", w.ID, "project_id", ws.ProjectID, "err", err.Error())
|
|
return row
|
|
}
|
|
row.OwnerID = pj.OwnerID
|
|
return row
|
|
}
|
|
|
|
// chatAttachmentAdapter implements runners.AttachmentCleanupRepo backed by
|
|
// chat.Repository. Translates chat.AttachmentRef → runners.AttachmentRow.
|
|
type chatAttachmentAdapter struct{ repo chat.Repository }
|
|
|
|
func (a chatAttachmentAdapter) ListAttachmentsForSoftDeletedConversations(ctx context.Context, olderThan time.Duration) ([]runners.AttachmentRow, error) {
|
|
refs, err := a.repo.ListAttachmentsForSoftDeletedConversations(ctx, olderThan)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return refsToRows(refs), nil
|
|
}
|
|
|
|
func (a chatAttachmentAdapter) ListExpiredOrphans(ctx context.Context, olderThan time.Duration) ([]runners.AttachmentRow, error) {
|
|
refs, err := a.repo.ListExpiredOrphans(ctx, olderThan)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return refsToRows(refs), nil
|
|
}
|
|
|
|
func (a chatAttachmentAdapter) DeleteAttachments(ctx context.Context, ids []uuid.UUID) error {
|
|
return a.repo.DeleteAttachments(ctx, ids)
|
|
}
|
|
|
|
func refsToRows(refs []chat.AttachmentRef) []runners.AttachmentRow {
|
|
out := make([]runners.AttachmentRow, 0, len(refs))
|
|
for _, r := range refs {
|
|
out = append(out, runners.AttachmentRow{ID: r.ID, StoragePath: r.StoragePath})
|
|
}
|
|
return out
|
|
}
|
|
|
|
// dispatcherAdapter implements runners.Dispatcher backed by *notify.Dispatcher.
|
|
type dispatcherAdapter struct{ d *notify.Dispatcher }
|
|
|
|
func (a dispatcherAdapter) Dispatch(ctx context.Context, msg notify.Message) error {
|
|
return a.d.Dispatch(ctx, msg)
|
|
}
|
|
|
|
// acpEventsPurgeAdapter exposes acp.Repository.PurgeEventsBefore as the narrow
|
|
// interface required by runners.AcpEventsPurge.
|
|
type acpEventsPurgeAdapter struct{ repo acp.Repository }
|
|
|
|
func (a acpEventsPurgeAdapter) PurgeEventsBefore(ctx context.Context, before time.Time) (int64, error) {
|
|
return a.repo.PurgeEventsBefore(ctx, before)
|
|
}
|