feat(jobs): WebhookNotifier enqueues webhook.deliver jobs (topic whitelist + disabled short-circuit)

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

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