You've already forked agentic-coding-workflow
113 lines
3.1 KiB
Go
113 lines
3.1 KiB
Go
// 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))
|
|
}
|