From 9f290335b2a0c9df13d4b8731af98831b35db4c7 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Tue, 5 May 2026 23:17:47 +0800 Subject: [PATCH] feat(jobs/handlers): webhook handler with HMAC sign + 4xx permanent / 5xx retry --- internal/jobs/handlers/webhook.go | 112 +++++++++++++++++++++++ internal/jobs/handlers/webhook_test.go | 121 +++++++++++++++++++++++++ 2 files changed, 233 insertions(+) create mode 100644 internal/jobs/handlers/webhook.go create mode 100644 internal/jobs/handlers/webhook_test.go diff --git a/internal/jobs/handlers/webhook.go b/internal/jobs/handlers/webhook.go new file mode 100644 index 0000000..e968aca --- /dev/null +++ b/internal/jobs/handlers/webhook.go @@ -0,0 +1,112 @@ +// Package handlers contains job.Handler implementations for the jobs module. +// Each file defines exactly one Handler so each can be tested in isolation. +package handlers + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "strconv" + "time" + + "github.com/yan1h/agent-coding-workflow/internal/infra/clock" + "github.com/yan1h/agent-coding-workflow/internal/jobs" +) + +// WebhookConfig is the per-handler webhook config (URL/secret/timeout). +// Topic-whitelist filtering is performed upstream by the WebhookNotifier +// before enqueue, so this handler always delivers what it gets. +type WebhookConfig struct { + URL string + Secret string + Timeout time.Duration +} + +// WebhookHandler implements jobs.Handler for the TypeWebhookDeliver type. +type WebhookHandler struct { + cfg WebhookConfig + clk clock.Clock + client *http.Client +} + +// NewWebhookHandler constructs a handler with the given config. The handler +// builds a fresh http.Client per dispatch (no keep-alive) since webhook +// frequency is low. +func NewWebhookHandler(cfg WebhookConfig, clk clock.Clock) *WebhookHandler { + if cfg.Timeout <= 0 { + cfg.Timeout = 10 * time.Second + } + if clk == nil { + clk = clock.Real() + } + return &WebhookHandler{ + cfg: cfg, + clk: clk, + client: &http.Client{Timeout: cfg.Timeout}, + } +} + +func (h *WebhookHandler) Type() jobs.JobType { return jobs.TypeWebhookDeliver } + +// payloadEnvelope mirrors the JSON shape the webhook receiver gets. Topic is +// surfaced as a header AND in the body for convenience. +type payloadEnvelope struct { + Topic string `json:"topic"` +} + +func (h *WebhookHandler) Handle(ctx context.Context, job jobs.Job) error { + if h.cfg.URL == "" { + return jobs.Permanent(fmt.Errorf("webhook url not configured")) + } + + var env payloadEnvelope + if err := json.Unmarshal(job.Payload, &env); err != nil { + return jobs.Permanent(fmt.Errorf("payload not valid json: %w", err)) + } + + body := []byte(job.Payload) + ts := strconv.FormatInt(h.clk.Now().Unix(), 10) + sig := sign(h.cfg.Secret, ts, body) + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, h.cfg.URL, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("build request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "agent-coding-workflow/dev") + req.Header.Set("X-ACW-Topic", env.Topic) + req.Header.Set("X-ACW-Event-Id", job.ID.String()) + req.Header.Set("X-ACW-Timestamp", ts) + req.Header.Set("X-ACW-Signature", sig) + + resp, err := h.client.Do(req) + if err != nil { + return fmt.Errorf("http: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + _, _ = io.Copy(io.Discard, resp.Body) + return nil + } + + respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 500)) + msg := fmt.Sprintf("%d %s", resp.StatusCode, string(respBody)) + if resp.StatusCode >= 400 && resp.StatusCode < 500 { + return jobs.Permanent(fmt.Errorf("%s", msg)) + } + return fmt.Errorf("%s", msg) +} + +func sign(secret, ts string, body []byte) string { + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write([]byte(ts + ".")) + mac.Write(body) + return "sha256=" + hex.EncodeToString(mac.Sum(nil)) +} diff --git a/internal/jobs/handlers/webhook_test.go b/internal/jobs/handlers/webhook_test.go new file mode 100644 index 0000000..2ae91d0 --- /dev/null +++ b/internal/jobs/handlers/webhook_test.go @@ -0,0 +1,121 @@ +package handlers_test + +import ( + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/yan1h/agent-coding-workflow/internal/infra/clock" + "github.com/yan1h/agent-coding-workflow/internal/jobs" + "github.com/yan1h/agent-coding-workflow/internal/jobs/handlers" +) + +func mkPayload(t *testing.T, topic string) json.RawMessage { + t.Helper() + p, err := json.Marshal(map[string]any{ + "topic": topic, "severity": "error", "title": "x", "body": "y", + "metadata": map[string]any{"k": "v"}, + "occurred_at": time.Now().UTC().Format(time.RFC3339), + }) + require.NoError(t, err) + return p +} + +func TestWebhook_2xxCompletes(t *testing.T) { + t.Parallel() + got := make(chan struct{}, 1) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got <- struct{}{} + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + h := handlers.NewWebhookHandler(handlers.WebhookConfig{ + URL: ts.URL, Secret: "s3cr3t", Timeout: time.Second, + }, clock.Real()) + err := h.Handle(context.Background(), jobs.Job{ + ID: uuid.New(), Type: jobs.TypeWebhookDeliver, Payload: mkPayload(t, "x"), + }) + require.NoError(t, err) + <-got +} + +func TestWebhook_4xxPermanent(t *testing.T) { + t.Parallel() + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + _, _ = w.Write([]byte("bad token")) + })) + defer ts.Close() + h := handlers.NewWebhookHandler(handlers.WebhookConfig{URL: ts.URL, Secret: "s", Timeout: time.Second}, clock.Real()) + err := h.Handle(context.Background(), jobs.Job{ID: uuid.New(), Type: jobs.TypeWebhookDeliver, Payload: mkPayload(t, "x")}) + require.Error(t, err) + assert.True(t, jobs.IsPermanent(err), "4xx should be permanent") + assert.Contains(t, err.Error(), "401") +} + +func TestWebhook_5xxTransient(t *testing.T) { + t.Parallel() + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer ts.Close() + h := handlers.NewWebhookHandler(handlers.WebhookConfig{URL: ts.URL, Secret: "s", Timeout: time.Second}, clock.Real()) + err := h.Handle(context.Background(), jobs.Job{ID: uuid.New(), Type: jobs.TypeWebhookDeliver, Payload: mkPayload(t, "x")}) + require.Error(t, err) + assert.False(t, jobs.IsPermanent(err)) +} + +func TestWebhook_HMACSignatureValidByReceiver(t *testing.T) { + t.Parallel() + const secret = "s3cr3t" + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + ts := r.Header.Get("X-ACW-Timestamp") + sig := r.Header.Get("X-ACW-Signature") + + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write([]byte(ts + "." + string(body))) + want := "sha256=" + hex.EncodeToString(mac.Sum(nil)) + if !hmac.Equal([]byte(want), []byte(sig)) { + http.Error(w, "bad sig", http.StatusUnauthorized) + return + } + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + h := handlers.NewWebhookHandler(handlers.WebhookConfig{URL: ts.URL, Secret: secret, Timeout: time.Second}, clock.Real()) + err := h.Handle(context.Background(), jobs.Job{ID: uuid.New(), Type: jobs.TypeWebhookDeliver, Payload: mkPayload(t, "x")}) + require.NoError(t, err) +} + +func TestWebhook_HeadersIncluded(t *testing.T) { + t.Parallel() + got := make(chan http.Header, 1) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got <- r.Header.Clone() + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + id := uuid.New() + h := handlers.NewWebhookHandler(handlers.WebhookConfig{URL: ts.URL, Secret: "s", Timeout: time.Second}, clock.Real()) + err := h.Handle(context.Background(), jobs.Job{ID: id, Type: jobs.TypeWebhookDeliver, Payload: mkPayload(t, "workspace.sync_failed")}) + require.NoError(t, err) + hdr := <-got + assert.Equal(t, "workspace.sync_failed", hdr.Get("X-ACW-Topic")) + assert.Equal(t, id.String(), hdr.Get("X-ACW-Event-Id")) + assert.True(t, strings.HasPrefix(hdr.Get("X-ACW-Signature"), "sha256=")) + assert.NotEmpty(t, hdr.Get("X-ACW-Timestamp")) +}