feat(app): wire jobs.Module + WebhookNotifier + integration test for webhook closed loop

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)
+}

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")
+	}
+}