From c7db7a9604e2d69d83734f1a83a772b393fbbb49 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Wed, 6 May 2026 09:25:19 +0800 Subject: [PATCH] feat(app): wire jobs.Module + WebhookNotifier + integration test for webhook closed loop --- internal/app/app.go | 320 ++++++++++++++++++++++++++++++- internal/app/integration_test.go | 92 +++++++++ 2 files changed, 411 insertions(+), 1 deletion(-) diff --git a/internal/app/app.go b/internal/app/app.go index 9b15988..1b5bda0 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -32,6 +32,7 @@ import ( "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/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" @@ -39,6 +40,9 @@ import ( "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" @@ -53,6 +57,8 @@ type App struct { pool *db.Pool server *http.Server dispatcher *notify.Dispatcher + jobs *jobs.Module + jobsCancel context.CancelFunc } // New 把所有模块按启动顺序拼装好。任何阶段失败都会返回错误，不留半成品资源。 @@ -231,6 +237,139 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) { chat.NewHandler(chatConvSvc, chatMsgSvc, chatTplSvc, chatEpSvc, chatUsageSvc, chatUploader, userSvc, userAdminAdapter{svc: userSvc}).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()) + 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) + 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) + + 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: time.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, + }, + }, 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, + }, + 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()) @@ -247,7 +386,7 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) { WriteTimeout: 60 * time.Second, IdleTimeout: 120 * time.Second, } - return &App{cfg: cfg, log: log, pool: pool, server: srv, dispatcher: notifyDispatcher}, nil + return &App{cfg: cfg, log: log, pool: pool, server: srv, dispatcher: notifyDispatcher, jobs: jobsModule, jobsCancel: jobsCancel}, nil } // Run 启动 HTTP server 并阻塞直到 ctx 取消（Shutdown）或服务器报错。 @@ -271,6 +410,14 @@ func (a *App) Run(ctx context.Context) error { if shutErr != nil { a.log.Error("server shutdown", "err", shutErr.Error()) } + // 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.jobsCancel() + stopCtx, stopCancel := context.WithTimeout(context.Background(), 30*time.Second) + a.jobs.Stop(stopCtx) + stopCancel() + } // server 已停止接收新请求；等异步通知投递结束再关 pool if err := a.dispatcher.Shutdown(shutCtx); err != nil { a.log.Warn("notify dispatcher drain", "err", err.Error()) @@ -280,6 +427,12 @@ func (a *App) Run(ctx context.Context) error { case err := <-errCh: drainCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() + if a.jobs != nil { + a.jobsCancel() + stopCtx, stopCancel := context.WithTimeout(context.Background(), 30*time.Second) + a.jobs.Stop(stopCtx) + stopCancel() + } _ = a.dispatcher.Shutdown(drainCtx) a.pool.Close() return err @@ -347,3 +500,168 @@ func projectACL(pj *project.Project, callerID uuid.UUID, isAdmin bool) (canRead, canRead = canWrite || pj.Visibility == project.VisibilityInternal return canRead, canWrite } + +// ===== 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, 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) +} diff --git a/internal/app/integration_test.go b/internal/app/integration_test.go index 211d39f..e181f75 100644 --- a/internal/app/integration_test.go +++ b/internal/app/integration_test.go @@ -27,11 +27,14 @@ import ( "github.com/google/uuid" "github.com/yan1h/agent-coding-workflow/internal/audit" + "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/logger" "github.com/yan1h/agent-coding-workflow/internal/infra/notify" + "github.com/yan1h/agent-coding-workflow/internal/jobs" + "github.com/yan1h/agent-coding-workflow/internal/jobs/handlers" "github.com/yan1h/agent-coding-workflow/internal/project" "github.com/yan1h/agent-coding-workflow/internal/transport/http/middleware" "github.com/yan1h/agent-coding-workflow/internal/user" @@ -499,3 +502,92 @@ func TestPlan3_WorkspaceClosedLoop(t *testing.T) { require.Equal(t, http.StatusNoContent, deleteJSON("/api/v1/worktrees/"+wtResp.ID.String()).StatusCode) require.Equal(t, http.StatusNoContent, deleteJSON("/api/v1/workspaces/"+wsResp.ID.String()).StatusCode) } + +func TestEndToEnd_WebhookClosedLoop(t *testing.T) { + ctx := context.Background() + + container, err := tcpg.Run(ctx, "postgres:16-alpine", + tcpg.WithDatabase("acw"), tcpg.WithUsername("acw"), tcpg.WithPassword("acw"), + ) + require.NoError(t, err) + t.Cleanup(func() { _ = container.Terminate(ctx) }) + + dsn, err := container.ConnectionString(ctx, "sslmode=disable") + require.NoError(t, err) + require.NoError(t, db.Migrate(ctx, dsn, "file://../../migrations")) + + pool, err := db.NewPool(ctx, dsn) + require.NoError(t, err) + t.Cleanup(pool.Close) + + log := logger.New(logger.Options{Format: logger.FormatJSON, Level: logger.LevelInfo}) + + // Webhook receiver + type recv struct { + body map[string]any + topic string + eventID string + sig string + } + received := make(chan recv, 4) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + rawBody, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(rawBody, &body) + received <- recv{ + body: body, + topic: r.Header.Get("X-ACW-Topic"), + eventID: r.Header.Get("X-ACW-Event-Id"), + sig: r.Header.Get("X-ACW-Signature"), + } + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(ts.Close) + + auditRec := audit.NewPostgresRecorder(pool) + notifyRepo := notify.NewPostgresRepository(pool) + disp := notify.NewDispatcher(notifyRepo, log) + + jobsRepo := jobs.NewPostgresRepository(pool) + h := handlers.NewWebhookHandler(handlers.WebhookConfig{ + URL: ts.URL, Secret: "test-secret", Timeout: 2 * time.Second, + }, clock.Real()) + n := jobs.NewWebhookNotifier(jobsRepo, jobs.WebhookNotifierConfig{ + Enabled: true, + Topics: []string{"workspace.sync_failed"}, + Backoff: []time.Duration{50 * time.Millisecond, 200 * time.Millisecond}, + }, clock.Real()) + disp.Register(n) + + // Run a single Worker (no full Module needed for this test). + w := jobs.NewWorker(jobsRepo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h}, + jobs.WorkerOptions{ + ID: "itest", PollInterval: 20 * time.Millisecond, + Backoff: []time.Duration{50 * time.Millisecond}, Clock: clock.Real(), Log: log, + }) + wctx, wcancel := context.WithCancel(ctx) + t.Cleanup(func() { wcancel(); w.Wait() }) + go w.Run(wctx) + + // Suppress unused-variable vet warning. + _ = auditRec + + eventTime := time.Now() + require.NoError(t, disp.Dispatch(ctx, notify.Message{ + ID: uuid.New(), UserID: uuid.New(), + Topic: "workspace.sync_failed", Severity: notify.SeverityError, + Title: "fetch failed", Body: "remote unreachable", + CreatedAt: eventTime, + })) + + select { + case got := <-received: + require.Equal(t, "workspace.sync_failed", got.topic) + require.NotEmpty(t, got.eventID) + require.Contains(t, got.sig, "sha256=") + require.Equal(t, "workspace.sync_failed", got.body["topic"]) + require.Equal(t, "fetch failed", got.body["title"]) + case <-time.After(3 * time.Second): + t.Fatal("webhook never fired") + } +}