From e3616705390ae9afe18800ca8b9550f94deaac37 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Tue, 5 May 2026 23:26:47 +0800 Subject: [PATCH] feat(jobs): WebhookNotifier enqueues webhook.deliver jobs (topic whitelist + disabled short-circuit) --- internal/jobs/notifier.go | 80 ++++++++++++++++++++++++++++++++ internal/jobs/notifier_test.go | 84 ++++++++++++++++++++++++++++++++++ 2 files changed, 164 insertions(+) create mode 100644 internal/jobs/notifier.go create mode 100644 internal/jobs/notifier_test.go diff --git a/internal/jobs/notifier.go b/internal/jobs/notifier.go new file mode 100644 index 0000000..13395d2 --- /dev/null +++ b/internal/jobs/notifier.go @@ -0,0 +1,80 @@ +package jobs + +import ( + "context" + "encoding/json" + "time" + + "github.com/yan1h/agent-coding-workflow/internal/infra/clock" + "github.com/yan1h/agent-coding-workflow/internal/infra/notify" +) + +// WebhookNotifierConfig drives the WebhookNotifier. Backoff defines the retry +// schedule passed to the worker via job MaxAttempts (len(Backoff)+1, since +// the first attempt is "attempt 1" not a retry). Topics is the whitelist of +// topics that get enqueued; everything else is dropped. +type WebhookNotifierConfig struct { + Enabled bool + Topics []string + Backoff []time.Duration +} + +// WebhookNotifier implements notify.Notifier. Its Send method does NOT actually +// deliver — it enqueues a webhook.deliver job. The Worker + WebhookHandler +// performs the real HTTP call. This indirection gives us retry / dead-letter / +// crash safety without coupling notify.Dispatcher to the jobs DB. +type WebhookNotifier struct { + repo Repository + cfg WebhookNotifierConfig + clk clock.Clock +} + +// NewWebhookNotifier constructs the notifier. +func NewWebhookNotifier(repo Repository, cfg WebhookNotifierConfig, clk clock.Clock) *WebhookNotifier { + if clk == nil { + clk = clock.Real() + } + return &WebhookNotifier{repo: repo, cfg: cfg, clk: clk} +} + +// Name reports the notify.Channel for this notifier. +func (n *WebhookNotifier) Name() notify.Channel { return notify.ChannelWebhook } + +// Send enqueues a webhook.deliver job. Disabled config / unmatched topics are +// silently skipped (returning nil, since the dispatcher's contract treats +// per-channel failures as non-fatal anyway). +func (n *WebhookNotifier) Send(ctx context.Context, msg notify.Message) error { + if !n.cfg.Enabled { + return nil + } + if !topicAllowed(n.cfg.Topics, msg.Topic) { + return nil + } + payload, err := json.Marshal(map[string]any{ + "topic": msg.Topic, + "severity": string(msg.Severity), + "title": msg.Title, + "body": msg.Body, + "link": msg.Link, + "metadata": msg.Metadata, + "occurred_at": msg.CreatedAt.UTC().Format(time.RFC3339), + }) + if err != nil { + return err + } + maxAttempts := int32(len(n.cfg.Backoff) + 1) + if maxAttempts < 1 { + maxAttempts = 1 + } + _, err = n.repo.Enqueue(ctx, TypeWebhookDeliver, payload, n.clk.Now(), maxAttempts) + return err +} + +func topicAllowed(allow []string, t string) bool { + for _, x := range allow { + if x == t { + return true + } + } + return false +} diff --git a/internal/jobs/notifier_test.go b/internal/jobs/notifier_test.go new file mode 100644 index 0000000..f9233a6 --- /dev/null +++ b/internal/jobs/notifier_test.go @@ -0,0 +1,84 @@ +//go:build integration + +package jobs_test + +import ( + "context" + "encoding/json" + "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/infra/notify" + "github.com/yan1h/agent-coding-workflow/internal/jobs" +) + +func TestWebhookNotifier_EnqueuesForWhitelistedTopic(t *testing.T) { + t.Parallel() + repo, _ := setupRepo(t) + ctx := context.Background() + n := jobs.NewWebhookNotifier(repo, jobs.WebhookNotifierConfig{ + Enabled: true, + Topics: []string{"workspace.sync_failed"}, + Backoff: []time.Duration{30 * time.Second}, + }, clock.Real()) + + err := n.Send(ctx, notify.Message{ + ID: uuid.New(), UserID: uuid.New(), Topic: "workspace.sync_failed", + Severity: notify.SeverityError, Title: "fail", Body: "x", + CreatedAt: time.Now(), + }) + require.NoError(t, err) + got, err := repo.Lease(ctx, "test-lease") + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, jobs.TypeWebhookDeliver, got.Type) + + var payload map[string]any + require.NoError(t, json.Unmarshal(got.Payload, &payload)) + assert.Equal(t, "workspace.sync_failed", payload["topic"]) +} + +func TestWebhookNotifier_DropsNonWhitelistedTopic(t *testing.T) { + t.Parallel() + repo, _ := setupRepo(t) + ctx := context.Background() + n := jobs.NewWebhookNotifier(repo, jobs.WebhookNotifierConfig{ + Enabled: true, + Topics: []string{"workspace.sync_failed"}, + Backoff: []time.Duration{30 * time.Second}, + }, clock.Real()) + + err := n.Send(ctx, notify.Message{ + ID: uuid.New(), UserID: uuid.New(), Topic: "chat.message_failed", + Severity: notify.SeverityError, CreatedAt: time.Now(), + }) + require.NoError(t, err) + got, _ := repo.Lease(ctx, "t") + assert.Nil(t, got, "non-whitelisted topic must not enqueue") +} + +func TestWebhookNotifier_DisabledShortCircuits(t *testing.T) { + t.Parallel() + repo, _ := setupRepo(t) + ctx := context.Background() + n := jobs.NewWebhookNotifier(repo, jobs.WebhookNotifierConfig{ + Enabled: false, + Topics: []string{"workspace.sync_failed"}, + }, clock.Real()) + err := n.Send(ctx, notify.Message{ + ID: uuid.New(), UserID: uuid.New(), Topic: "workspace.sync_failed", CreatedAt: time.Now(), + }) + require.NoError(t, err) + got, _ := repo.Lease(ctx, "t") + assert.Nil(t, got) +} + +func TestWebhookNotifier_Name(t *testing.T) { + n := jobs.NewWebhookNotifier(nil, jobs.WebhookNotifierConfig{}, clock.Real()) + assert.Equal(t, notify.ChannelWebhook, n.Name()) +}