agentic-coding-workflow/internal/app/integration_test.go

//go:build integration

// Package app_test 中的 integration_test.go 端到端覆盖：bootstrap admin →
// 通过 HTTP 登录拿到 token → dispatcher 写一条通知 → 用 token 通过 HTTP
// 查询未读计数为 1。所有外部依赖只有一个 postgres 容器，由 testcontainers
// 启动。需要本机 Docker；跑命令：
//
//	go test -tags=integration -count=1 ./internal/app/...
package app_test

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"testing"
	"time"

	"github.com/go-chi/chi/v5"
	"github.com/stretchr/testify/require"
	tcpg "github.com/testcontainers/testcontainers-go/modules/postgres"

	"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"
	"github.com/yan1h/agent-coding-workflow/internal/workspace"
)

func TestEndToEnd_LoginAndDispatchNotification(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})
	auditRec := audit.NewPostgresRecorder(pool)

	userRepo := user.NewPostgresRepository(pool)
	userSvc := user.NewService(userRepo, auditRec)

	notifyRepo := notify.NewPostgresRepository(pool)
	disp := notify.NewDispatcher(notifyRepo, log)
	disp.Register(notify.NewDummyNotifier(log))

	u, err := userSvc.Bootstrap(ctx, "admin@local", "password123")
	require.NoError(t, err)
	require.NotNil(t, u)

	r := chi.NewRouter()
	r.Use(middleware.RequestID)
	user.NewHandler(userSvc).Mount(r)
	// userSvc 直接满足 middleware.SessionResolver；不需要任何函数适配器。
	notify.NewHandler(notifyRepo, userSvc).Mount(r)

	srv := httptest.NewServer(r)
	t.Cleanup(srv.Close)

	body, _ := json.Marshal(map[string]string{"email": "admin@local", "password": "password123"})
	resp, err := http.Post(srv.URL+"/api/v1/auth/login", "application/json", bytes.NewReader(body))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, resp.StatusCode)

	var lr struct {
		Token string `json:"token"`
	}
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&lr))
	_ = resp.Body.Close()
	require.NotEmpty(t, lr.Token)

	require.NoError(t, disp.Dispatch(ctx, notify.Message{
		UserID:   u.ID,
		Topic:    "test",
		Severity: notify.SeverityInfo,
		Title:    "hello",
		Body:     "world",
	}))

	require.Eventually(t, func() bool {
		req, _ := http.NewRequest("GET", srv.URL+"/api/v1/notifications/unread-count", nil)
		req.Header.Set("Authorization", "Bearer "+lr.Token)
		r2, err := http.DefaultClient.Do(req)
		if err != nil {
			return false
		}
		defer r2.Body.Close()
		var out struct {
			Count int `json:"count"`
		}
		_ = json.NewDecoder(r2.Body).Decode(&out)
		return out.Count == 1
	}, 2*time.Second, 50*time.Millisecond, "expected unread count to reach 1 after dispatch")
}

// userAdminAdapterTest 把 user.Service 适配为 project.AdminLookup（test-internal）。
type userAdminAdapterTest struct{ svc user.Service }

func (a userAdminAdapterTest) 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
}

func TestEndToEnd_ProjectClosedLoop(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)

	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, nil, nil)
	issueSvc := project.NewIssueService(projectRepo, auditRec)
	artifactSvc := project.NewArtifactService(projectRepo, auditRec)

	adminUser, err := userSvc.Bootstrap(ctx, "admin@local", "password123")
	require.NoError(t, err)
	require.NotNil(t, adminUser)

	r := chi.NewRouter()
	r.Use(middleware.RequestID)
	user.NewHandler(userSvc).Mount(r)
	project.NewHandler(projectSvc, requirementSvc, issueSvc, artifactSvc, userSvc, userAdminAdapterTest{svc: userSvc}).Mount(r)

	srv := httptest.NewServer(r)
	t.Cleanup(srv.Close)

	// 登录拿 token
	loginBody, _ := json.Marshal(map[string]string{"email": "admin@local", "password": "password123"})
	loginResp, err := http.Post(srv.URL+"/api/v1/auth/login", "application/json", bytes.NewReader(loginBody))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, loginResp.StatusCode)

	var lr struct {
		Token string `json:"token"`
	}
	require.NoError(t, json.NewDecoder(loginResp.Body).Decode(&lr))
	_ = loginResp.Body.Close()
	require.NotEmpty(t, lr.Token)

	// post 辅助函数：发送带 token 的 POST 请求，返回状态码。
	post := func(t *testing.T, path string, body any) int {
		t.Helper()
		b, _ := json.Marshal(body)
		req, _ := http.NewRequestWithContext(ctx, http.MethodPost, srv.URL+path, bytes.NewReader(b))
		req.Header.Set("Content-Type", "application/json")
		req.Header.Set("Authorization", "Bearer "+lr.Token)
		resp, err := http.DefaultClient.Do(req)
		require.NoError(t, err)
		_ = resp.Body.Close()
		return resp.StatusCode
	}

	// PM 闭环验证
	require.Equal(t, http.StatusCreated, post(t, "/api/v1/projects/", map[string]string{"slug": "demo", "name": "Demo"}))
	require.Equal(t, http.StatusConflict, post(t, "/api/v1/projects/", map[string]string{"slug": "demo", "name": "Demo"}))
	require.Equal(t, http.StatusCreated, post(t, "/api/v1/projects/demo/requirements", map[string]string{"title": "Req1"}))
	require.Equal(t, http.StatusCreated, post(t, "/api/v1/projects/demo/requirements", map[string]string{"title": "Req2"}))
	require.Equal(t, http.StatusOK, post(t, "/api/v1/projects/demo/requirements/1/phase", map[string]string{"phase": "implementing"}))
	require.Equal(t, http.StatusCreated, post(t, "/api/v1/projects/demo/issues", map[string]any{"title": "I1", "requirement_number": 1}))
	require.Equal(t, http.StatusCreated, post(t, "/api/v1/projects/demo/issues", map[string]string{"title": "Free"}))
	require.Equal(t, http.StatusNoContent, post(t, "/api/v1/projects/demo/requirements/1/close", nil))
	require.Equal(t, http.StatusConflict, post(t, "/api/v1/projects/demo/requirements/1/phase", map[string]string{"phase": "done"}))
	require.Equal(t, http.StatusNoContent, post(t, "/api/v1/projects/demo/archive", nil))
	require.Equal(t, http.StatusConflict, post(t, "/api/v1/projects/demo/requirements", map[string]string{"title": "Should fail"}))

	// 验证 audit_logs 中 project/requirement/issue 相关行数 >= 8
	var count int
	err = pool.QueryRow(ctx,
		`SELECT COUNT(*) FROM audit_logs WHERE action LIKE 'project.%' OR action LIKE 'requirement.%' OR action LIKE 'issue.%'`,
	).Scan(&count)
	require.NoError(t, err)
	require.GreaterOrEqual(t, count, 8, "expected at least 8 audit log entries for project/requirement/issue actions")
}

// projectAccessAdapterTest 把 project.ProjectService.Get / GetByID 桥接成 workspace.ProjectAccess。
// 与 internal/app/app.go 里的 projectAccessAdapter 同形（拷贝粘贴），保持 test-internal 不导出。
type projectAccessAdapterTest struct {
	p project.ProjectService
}

func (a projectAccessAdapterTest) 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 := projectACLTest(pj, callerID, isAdmin)
	return pj.ID, canRead, canWrite, nil
}

func (a projectAccessAdapterTest) 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 := projectACLTest(pj, callerID, isAdmin)
	return canRead, canWrite, nil
}

func projectACLTest(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
}

// makeLocalBareRepo 在临时目录里建一个带 main 分支与一次 commit 的 bare 仓库，
// 返回路径（用 file:// 协议作为 git_remote_url，让 Plan #3 的 clone 走 file 协议）。
func makeLocalBareRepo(t *testing.T) string {
	t.Helper()
	dir := t.TempDir()
	bare := filepath.Join(dir, "remote.git")
	require.NoError(t, exec.Command("git", "init", "--bare", "-b", "main", bare).Run())
	work := filepath.Join(dir, "seed")
	require.NoError(t, os.MkdirAll(work, 0o755))
	for _, args := range [][]string{
		{"init", "-b", "main"},
		{"config", "user.email", "t@e.c"},
		{"config", "user.name", "T"},
	} {
		c := exec.Command("git", args...)
		c.Dir = work
		require.NoError(t, c.Run())
	}
	require.NoError(t, os.WriteFile(filepath.Join(work, "README.md"), []byte("hi\n"), 0o644))
	for _, args := range [][]string{
		{"add", "."},
		{"commit", "-m", "init"},
		{"remote", "add", "origin", bare},
		{"push", "origin", "main"},
	} {
		c := exec.Command("git", args...)
		c.Dir = work
		require.NoError(t, c.Run())
	}
	return bare
}

// TestPlan3_WorkspaceClosedLoop 端到端覆盖 workspace 模块：
//  1. bootstrap admin → login 拿 token
//  2. 创建 project + bare 远端
//  3. POST /workspaces 创建 workspace（异步 clone）
//  4. 轮询 GET /workspaces/{id} 直到 sync_status == "idle"
//  5. POST /worktrees 创建支线 → acquire/release 状态机切换
//  6. POST /sync（等同 git fetch）→ sync_status 仍为 idle
//  7. DELETE worktree → DELETE workspace 都返回 204
//
// 跳过条件：本机没有 git 二进制时直接 t.Skip。Docker 不可用会让 testcontainers
// Run 失败，由上层的 require 触发测试失败（与已有 PM 闭环测试一致）。
func TestPlan3_WorkspaceClosedLoop(t *testing.T) {
	if _, err := exec.LookPath("git"); err != nil {
		t.Skip("git not available")
	}
	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})
	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, nil, nil)
	issueSvc := project.NewIssueService(projectRepo, auditRec)
	artifactSvc := project.NewArtifactService(projectRepo, auditRec)

	adminUser, err := userSvc.Bootstrap(ctx, "admin@local", "password123")
	require.NoError(t, err)
	require.NotNil(t, adminUser)

	// ---- workspace 模块装配（与 app.go 同形，但不依赖 cfg） ----
	dataDir := t.TempDir()
	tmpDir := filepath.Join(dataDir, "tmp")
	require.NoError(t, os.MkdirAll(tmpDir, 0o700))

	enc, err := crypto.NewEncryptor(make([]byte, 32))
	require.NoError(t, err)

	gitr := git.NewDefaultRunner(git.Config{
		Binary: "git", TmpDir: tmpDir,
		CloneTimeout: 30 * time.Second, FetchTimeout: 30 * time.Second,
		PushTimeout: 30 * time.Second, OpsTimeout: 30 * time.Second,
	})

	wsRepo := workspace.NewPostgresRepository(pool)
	_, err = wsRepo.ResetStuckSyncStatuses(ctx)
	require.NoError(t, err)

	pa := projectAccessAdapterTest{p: projectSvc}
	wsSvc := workspace.NewWorkspaceService(wsRepo, auditRec, enc, gitr, pa, nil, dataDir)

	credAcc, ok := wsSvc.(workspace.CredentialAccessor)
	require.True(t, ok)
	credLoader := func(ctx context.Context, wsID uuid.UUID) (*git.Credential, error) {
		return credAcc.LoadDecryptedCredential(ctx, wsID)
	}
	cloneRunner := workspace.NewCloneRunner(wsRepo, auditRec, gitr, log, credLoader, 30*time.Second)
	if setter, ok := wsSvc.(workspace.SchedulerSetter); ok {
		setter.SetScheduler(cloneRunner)
	}
	wtSvc := workspace.NewWorktreeService(wsRepo, auditRec, gitr, pa, dataDir)
	gopSvc := workspace.NewGitOpsService(wsRepo, auditRec, gitr, pa, credLoader)

	r := chi.NewRouter()
	r.Use(middleware.RequestID)
	user.NewHandler(userSvc).Mount(r)
	project.NewHandler(projectSvc, requirementSvc, issueSvc, artifactSvc, userSvc, userAdminAdapterTest{svc: userSvc}).Mount(r)
	workspace.NewHandler(wsSvc, wtSvc, gopSvc, userSvc, userAdminAdapterTest{svc: userSvc}).Mount(r)

	srv := httptest.NewServer(r)
	t.Cleanup(srv.Close)

	// ---- 登录拿 token ----
	loginBody, _ := json.Marshal(map[string]string{"email": "admin@local", "password": "password123"})
	loginResp, err := http.Post(srv.URL+"/api/v1/auth/login", "application/json", bytes.NewReader(loginBody))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, loginResp.StatusCode)
	var lr struct {
		Token string `json:"token"`
	}
	require.NoError(t, json.NewDecoder(loginResp.Body).Decode(&lr))
	_ = loginResp.Body.Close()
	require.NotEmpty(t, lr.Token)
	token := lr.Token

	// 三个内联 helper：内联是有意为之，避免引入 test 外部 helper 包。
	postJSON := func(path string, body any) (*http.Response, []byte) {
		t.Helper()
		b, _ := json.Marshal(body)
		req, _ := http.NewRequestWithContext(ctx, http.MethodPost, srv.URL+path, bytes.NewReader(b))
		req.Header.Set("Content-Type", "application/json")
		req.Header.Set("Authorization", "Bearer "+token)
		resp, err := http.DefaultClient.Do(req)
		require.NoError(t, err)
		buf, _ := io.ReadAll(resp.Body)
		_ = resp.Body.Close()
		return resp, buf
	}
	getJSON := func(path string) (*http.Response, []byte) {
		t.Helper()
		req, _ := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+path, nil)
		req.Header.Set("Authorization", "Bearer "+token)
		resp, err := http.DefaultClient.Do(req)
		require.NoError(t, err)
		buf, _ := io.ReadAll(resp.Body)
		_ = resp.Body.Close()
		return resp, buf
	}
	deleteJSON := func(path string) *http.Response {
		t.Helper()
		req, _ := http.NewRequestWithContext(ctx, http.MethodDelete, srv.URL+path, nil)
		req.Header.Set("Authorization", "Bearer "+token)
		resp, err := http.DefaultClient.Do(req)
		require.NoError(t, err)
		_ = resp.Body.Close()
		return resp
	}

	// ---- 1. 创建 project ----
	resp, _ := postJSON("/api/v1/projects/", map[string]string{"slug": "p1", "name": "P1"})
	require.Equal(t, http.StatusCreated, resp.StatusCode)

	// ---- 2. 准备本地 bare 仓库（file://协议） ----
	bare := makeLocalBareRepo(t)
	remoteURL := "file://" + filepath.ToSlash(bare)

	// ---- 3. 创建 workspace（credential.kind=none，本地 file 协议无需凭据） ----
	resp, body := postJSON("/api/v1/projects/p1/workspaces", map[string]any{
		"slug": "ws1", "name": "WS1",
		"git_remote_url": remoteURL,
		"default_branch": "main",
		"credential":     map[string]any{"kind": "none"},
	})
	require.Equalf(t, http.StatusCreated, resp.StatusCode, "create workspace failed: %s", string(body))
	var wsResp struct {
		ID         uuid.UUID `json:"id"`
		SyncStatus string    `json:"sync_status"`
	}
	require.NoError(t, json.Unmarshal(body, &wsResp))
	require.NotEqual(t, uuid.Nil, wsResp.ID)

	// ---- 4. 等 clone 完成 → sync_status = idle ----
	require.Eventuallyf(t, func() bool {
		r, b := getJSON("/api/v1/workspaces/" + wsResp.ID.String())
		if r.StatusCode != http.StatusOK {
			return false
		}
		var ws struct {
			SyncStatus    string `json:"sync_status"`
			LastSyncError string `json:"last_sync_error"`
		}
		_ = json.Unmarshal(b, &ws)
		if ws.SyncStatus == "error" {
			t.Logf("clone errored: %s", ws.LastSyncError)
			return false
		}
		return ws.SyncStatus == "idle"
	}, 30*time.Second, 200*time.Millisecond, "workspace clone did not become idle")

	// ---- 5. 创建 worktree feat-x ----
	resp, body = postJSON("/api/v1/workspaces/"+wsResp.ID.String()+"/worktrees", map[string]string{
		"branch": "feat-x", "base_branch": "main",
	})
	require.Equalf(t, http.StatusCreated, resp.StatusCode, "create worktree: %s", string(body))
	var wtResp struct {
		ID     uuid.UUID `json:"id"`
		Branch string    `json:"branch"`
		Status string    `json:"status"`
	}
	require.NoError(t, json.Unmarshal(body, &wtResp))
	require.Equal(t, "feat-x", wtResp.Branch)
	require.Equal(t, "idle", wtResp.Status)

	// ---- 6. acquire / release ----
	resp, body = postJSON("/api/v1/worktrees/"+wtResp.ID.String()+"/acquire", nil)
	require.Equalf(t, http.StatusOK, resp.StatusCode, "acquire: %s", string(body))
	var afterAcq struct {
		Status string `json:"status"`
	}
	require.NoError(t, json.Unmarshal(body, &afterAcq))
	require.Equal(t, "active", afterAcq.Status)

	resp, body = postJSON("/api/v1/worktrees/"+wtResp.ID.String()+"/release", nil)
	require.Equalf(t, http.StatusOK, resp.StatusCode, "release: %s", string(body))
	var afterRel struct {
		Status string `json:"status"`
	}
	require.NoError(t, json.Unmarshal(body, &afterRel))
	require.Equal(t, "idle", afterRel.Status)

	// ---- 7. 同步 workspace（git fetch）→ 仍为 idle ----
	resp, body = postJSON("/api/v1/workspaces/"+wsResp.ID.String()+"/sync", nil)
	require.Equalf(t, http.StatusOK, resp.StatusCode, "sync: %s", string(body))
	var afterSync struct {
		SyncStatus string `json:"sync_status"`
	}
	require.NoError(t, json.Unmarshal(body, &afterSync))
	require.Equal(t, "idle", afterSync.SyncStatus)

	// ---- 8. 删除 worktree / workspace ----
	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(), auditRec, log)
	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)

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