feat: implement webhook sender worker and update push delivery dispatcher (#2306)

- Created the core webhook sender logic, including URL validation (restricted to https://hooks.slack.com/) and Markdown payload generation for Slack.
- Added tests validating realistic search query formats and proper URL encoding.
- Updated the push delivery dispatcher fan-out logic to publish to the webhook queue when matching triggers are found.

Author: neilv-g

backend/pkg/httpserver/create_notification_channel.go

@@ -23,6 +23,7 @@ import (
 	"github.com/GoogleChrome/webstatus.dev/lib/backendtypes"
 	"github.com/GoogleChrome/webstatus.dev/lib/gen/openapi/backend"
 	"github.com/GoogleChrome/webstatus.dev/lib/httpmiddlewares"
+	"github.com/GoogleChrome/webstatus.dev/lib/httputils"
 )
 
 func validateNotificationChannel(input *backend.CreateNotificationChannelRequest) *fieldValidationErrors {
@@ -33,7 +34,7 @@ func validateNotificationChannel(input *backend.CreateNotificationChannelRequest
 	}
 
 	if cfg, err := input.Config.AsWebhookConfig(); err == nil && cfg.Type == backend.WebhookConfigTypeWebhook {
-		if err := validateSlackWebhookURL(cfg.Url); err != nil {
+		if err := httputils.ValidateSlackWebhookURL(cfg.Url); err != nil {
 			fieldErrors.addFieldError("config.url", err)
 		}
 	} else {

backend/pkg/httpserver/notification_validation.go

@@ -15,10 +15,7 @@
 package httpserver
 
 import (
-	"errors"
 	"fmt"
-	"net/url"
-	"strings"
 )
 
 const (
@@ -27,26 +24,6 @@ const (
 )
 
 var (
-	errInvalidSlackWebhookURL = errors.New(
-		"invalid Slack webhook URL. Must be a valid https://hooks.slack.com/services/ URL")
 	errNotificationChannelInvalidNameLength = fmt.Errorf("name must be between %d and %d characters long",
 		notificationChannelNameMinLength, notificationChannelNameMaxLength)
 )
-
-// Validates the URL matches the expected Slack webhook URL format as defined by
-// https://docs.slack.dev/messaging/sending-messages-using-incoming-webhooks/
-// Ex. "https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX"
-func validateSlackWebhookURL(webhookURL string) error {
-	u, err := url.Parse(webhookURL)
-	if err != nil {
-		return err
-	}
-
-	if u.Scheme != "https" ||
-		u.Host != "hooks.slack.com" ||
-		!strings.HasPrefix(u.Path, "/services/") {
-		return errInvalidSlackWebhookURL
-	}
-
-	return nil
-}

backend/pkg/httpserver/update_notification_channel.go

@@ -23,6 +23,7 @@ import (
 	"github.com/GoogleChrome/webstatus.dev/lib/backendtypes"
 	"github.com/GoogleChrome/webstatus.dev/lib/gen/openapi/backend"
 	"github.com/GoogleChrome/webstatus.dev/lib/httpmiddlewares"
+	"github.com/GoogleChrome/webstatus.dev/lib/httputils"
 )
 
 func validateUpdateNotificationChannel(request *backend.UpdateNotificationChannelRequest) *fieldValidationErrors {
@@ -44,7 +45,7 @@ func validateUpdateNotificationChannel(request *backend.UpdateNotificationChanne
 			}
 
 			if cfg, err := request.Config.AsWebhookConfig(); err == nil && cfg.Type == backend.WebhookConfigTypeWebhook {
-				if err := validateSlackWebhookURL(cfg.Url); err != nil {
+				if err := httputils.ValidateSlackWebhookURL(cfg.Url); err != nil {
 					fieldErrors.addFieldError("config.url", err)
 				}
 			} else {

lib/httputils/slack_webhook.go

@@ -0,0 +1,45 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package httputils
+
+import (
+	"errors"
+	"net/url"
+	"strings"
+)
+
+var (
+	// ErrInvalidSlackWebhookURL is returned when the Slack webhook URL is invalid.
+	ErrInvalidSlackWebhookURL = errors.New(
+		"invalid Slack webhook URL. Must be a valid https://hooks.slack.com/services/ URL")
+)
+
+// ValidateSlackWebhookURL validates the URL matches the expected Slack webhook URL format as defined by
+// https://docs.slack.dev/messaging/sending-messages-using-incoming-webhooks/
+// Ex. "https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX"
+func ValidateSlackWebhookURL(webhookURL string) error {
+	u, err := url.Parse(webhookURL)
+	if err != nil {
+		return err
+	}
+
+	if u.Scheme != "https" ||
+		u.Host != "hooks.slack.com" ||
+		!strings.HasPrefix(u.Path, "/services/") {
+		return ErrInvalidSlackWebhookURL
+	}
+
+	return nil
+}

…tpserver/notification_validation_test.go lib/httputils/slack_webhook_test.gobackend/pkg/httpserver/notification_validation_test.go renamed to lib/httputils/slack_webhook_test.go backend/pkg/httpserver/notification_validation_test.go renamed to lib/httputils/slack_webhook_test.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package httpserver
+package httputils
 
 import (
 	"errors"
@@ -34,17 +34,17 @@ func TestValidateSlackWebhookURL(t *testing.T) {
 		{
 			name:          "Invalid Scheme (http)",
 			url:           "http://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX",
-			expectedError: errInvalidSlackWebhookURL,
+			expectedError: ErrInvalidSlackWebhookURL,
 		},
 		{
 			name:          "Invalid Host",
 			url:           "https://slack.hooks.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX",
-			expectedError: errInvalidSlackWebhookURL,
+			expectedError: ErrInvalidSlackWebhookURL,
 		},
 		{
 			name:          "Invalid Path (missing /services/)",
 			url:           "https://hooks.slack.com/foo/bar",
-			expectedError: errInvalidSlackWebhookURL,
+			expectedError: ErrInvalidSlackWebhookURL,
 		},
 		{
 			name:          "Invalid URL format",
@@ -55,7 +55,7 @@ func TestValidateSlackWebhookURL(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			err := validateSlackWebhookURL(tc.url)
+			err := ValidateSlackWebhookURL(tc.url)
 
 			if tc.name == "Invalid URL format" {
 				if err == nil {

workers/push_delivery/cmd/job/main.go

@@ -65,6 +65,7 @@ func main() {
 		slog.ErrorContext(ctx, "EMAIL_TOPIC_ID is not set. exiting...")
 		os.Exit(1)
 	}
+	// Push destination 2: Webhook
 	webhookTopicID := os.Getenv("WEBHOOK_TOPIC_ID")
 	if webhookTopicID == "" {
 		slog.ErrorContext(ctx, "WEBHOOK_TOPIC_ID is not set. exiting...")

workers/push_delivery/pkg/dispatcher/dispatcher.go

@@ -31,6 +31,7 @@ type SubscriptionFinder interface {
 
 type DeliveryPublisher interface {
 	PublishEmailJob(ctx context.Context, job workertypes.EmailDeliveryJob) error
+	PublishWebhookJob(ctx context.Context, job workertypes.WebhookDeliveryJob) error
 }
 
 // SummaryParser abstracts the logic for parsing the event summary blob.
@@ -56,17 +57,19 @@ func (d *Dispatcher) ProcessEvent(ctx context.Context,
 	metadata workertypes.DispatchEventMetadata, summary []byte) error {
 	slog.InfoContext(ctx, "processing event", "event_id", metadata.EventID, "search_id", metadata.SearchID)
 
-	// 1. Generate Delivery Jobs from Event Summary
+	// 1. Generate Delivery Jobs from Event Summary.
 	gen := &deliveryJobGenerator{
 		finder:   d.finder,
 		metadata: metadata,
 		// We pass the raw summary bytes down so it can be attached to the jobs
 		// without needing to re-marshal the struct.
-		rawSummary: summary,
-		emailJobs:  nil,
+		rawSummary:  summary,
+		emailJobs:   nil,
+		webhookJobs: nil,
 	}
 
 	if err := d.parser(gen.rawSummary, gen); err != nil {
+
 		return fmt.Errorf("failed to parse event summary: %w", err)
 	}
 
@@ -79,11 +82,11 @@ func (d *Dispatcher) ProcessEvent(ctx context.Context,
 
 	slog.InfoContext(ctx, "dispatching jobs", "count", totalJobs)
 
-	// 2. Publish Delivery Jobs
+	// 2. Publish Delivery Jobs.
 	successCount := 0
 	failCount := 0
 
-	// Publish Email Jobs
+	// Publish Email Jobs.
 	for _, job := range gen.emailJobs {
 		if err := d.publisher.PublishEmailJob(ctx, job); err != nil {
 			slog.ErrorContext(ctx, "failed to publish email job",
@@ -94,8 +97,16 @@ func (d *Dispatcher) ProcessEvent(ctx context.Context,
 		}
 	}
 
-	// TODO: Webhook jobs would be published here similarly
-	// https://github.com/GoogleChrome/webstatus.dev/issues/1859
+	// Publish Webhook Jobs.
+	for _, job := range gen.webhookJobs {
+		if err := d.publisher.PublishWebhookJob(ctx, job); err != nil {
+			slog.ErrorContext(ctx, "failed to publish webhook job",
+				"subscription_id", job.SubscriptionID, "error", err)
+			failCount++
+		} else {
+			successCount++
+		}
+	}
 
 	slog.InfoContext(ctx, "dispatch complete",
 		"event_id", metadata.EventID,
@@ -112,10 +123,11 @@ func (d *Dispatcher) ProcessEvent(ctx context.Context,
 
 // deliveryJobGenerator implements workertypes.SummaryVisitor to generate jobs from V1 summaries.
 type deliveryJobGenerator struct {
-	finder     SubscriptionFinder
-	metadata   workertypes.DispatchEventMetadata
-	rawSummary []byte
-	emailJobs  []workertypes.EmailDeliveryJob
+	finder      SubscriptionFinder
+	metadata    workertypes.DispatchEventMetadata
+	rawSummary  []byte
+	emailJobs   []workertypes.EmailDeliveryJob
+	webhookJobs []workertypes.WebhookDeliveryJob
 }
 
 func (g *deliveryJobGenerator) VisitV1(s workertypes.EventSummary) error {
@@ -127,6 +139,7 @@ func (g *deliveryJobGenerator) VisitV1(s workertypes.EventSummary) error {
 		g.metadata.SearchID,
 		g.metadata.Frequency)
 	if err != nil {
+
 		return fmt.Errorf("failed to find subscribers: %w", err)
 	}
 
@@ -144,7 +157,7 @@ func (g *deliveryJobGenerator) VisitV1(s workertypes.EventSummary) error {
 	}
 
 	// 2. Filter & Create Jobs
-	// Iterate Emails
+	// Iterate Emails.
 	for _, sub := range subscribers.Emails {
 		if !shouldNotifyV1(sub.Triggers, s) {
 			continue
@@ -159,17 +172,28 @@ func (g *deliveryJobGenerator) VisitV1(s workertypes.EventSummary) error {
 		})
 	}
 
-	// TODO: Iterate Webhooks when supported.
-	// https://github.com/GoogleChrome/webstatus.dev/issues/1859
+	// Iterate Webhooks.
+	for _, sub := range subscribers.Webhooks {
+		if !shouldNotifyV1(sub.Triggers, s) {
+			continue
+		}
+		g.webhookJobs = append(g.webhookJobs, workertypes.WebhookDeliveryJob{
+			SubscriptionID: sub.SubscriptionID,
+			WebhookURL:     sub.WebhookURL,
+			WebhookType:    sub.WebhookType,
+			SummaryRaw:     g.rawSummary,
+			Metadata:       deliveryMetadata,
+			ChannelID:      sub.ChannelID,
+			Triggers:       sub.Triggers,
+		})
+	}
 
 	return nil
 }
 
 // JobCount returns the total number of delivery jobs generated.
 func (g *deliveryJobGenerator) JobCount() int {
-	// TODO: When we add Webhook jobs, sum them here too.
-
-	return len(g.emailJobs)
+	return len(g.emailJobs) + len(g.webhookJobs)
 }
 
 // shouldNotifyV1 determines if the V1 event summary matches any of the user's triggers.