feat(jobs/handlers): webhook handler with HMAC sign + 4xx permanent / 5xx retry

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

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