feat(push_delivery,backend): cut over push delivery and RSS handler to BaseSummaryVisitor (#2622) (#2637)

* feat(push_delivery,backend): cut over push delivery and RSS handler to BaseSummaryVisitor (#2622)

Complete the Visitor pattern cutover across push delivery pipelines and the RSS HTTP handler.

- Update shouldNotifyV1 in workers/push_delivery/pkg/dispatcher to delegate highlight filtering and error checks to EventSummary.Categorize(triggers) and BaseSummaryVisitor.HasContent().
- Cut over get_subscription_rss.go in backend/pkg/httpserver to invoke summary.Accept(visitor, triggers) for RSS feed payload generation.
- Pass EventSummary by pointer (*EventSummary) into shouldNotifyV1 to avoid header and slice backing array copy overhead across subscriber iteration loops.
- Remove redundant manual trigger matching logic, completing the double-dispatch cutover for push notifications and RSS feed rendering.

CONV=f70d8c59-6f49-4a69-bb74-5643e908f5b1
TAG=agy

* address feedback

Author: jcscottiii

backend/pkg/httpserver/get_subscription_rss.go

@@ -17,6 +17,7 @@ package httpserver
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"encoding/xml"
 	"errors"
 	"fmt"
@@ -176,22 +177,11 @@ func (s *Server) GetSubscriptionRSS(
 		})
 	}
 
-	var jobTriggers []workertypes.JobTrigger
-	for _, triggerItem := range sub.Triggers {
-		triggerVal, err := triggerItem.Value.AsSubscriptionTriggerWritable()
-		if err != nil {
-			continue
-		}
-		if jobTrigger, ok := workertypes.ToJobTrigger(triggerVal); ok {
-			jobTriggers = append(jobTriggers, jobTrigger)
-		}
-	}
+	jobTriggers := extractJobTriggers(sub.Triggers)
 
 	for _, e := range events {
-		visitor := newRSSVisitor(jobTriggers)
-		var description string
-		var title string
-		if err := workertypes.ParseEventSummary(e.Summary, visitor); err != nil {
+		var summary workertypes.EventSummary
+		if err := json.Unmarshal([]byte(e.Summary), &summary); err != nil {
 			slog.ErrorContext(ctx, "failed to unmarshal summary", "event_id", e.ID, "error", err)
 
 			errorHTML := fmt.Sprintf(
@@ -212,6 +202,16 @@ func (s *Server) GetSubscriptionRSS(
 			continue
 		}
 
+		var description string
+		var title string
+
+		visitor := newRSSVisitor(jobTriggers)
+		if err := visitor.VisitV1(summary); err != nil {
+			slog.ErrorContext(ctx, "failed to process RSS summary visitor", "event_id", e.ID, "error", err)
+
+			continue
+		}
+
 		if !visitor.HasContent() {
 			continue
 		}
@@ -262,3 +262,18 @@ func (s *Server) GetSubscriptionRSS(
 		ContentLength: int64(buf.Len()),
 	}, nil
 }
+
+func extractJobTriggers(triggers []backend.SubscriptionTriggerResponseItem) []workertypes.JobTrigger {
+	var jobTriggers []workertypes.JobTrigger
+	for _, triggerItem := range triggers {
+		triggerVal, err := triggerItem.Value.AsSubscriptionTriggerWritable()
+		if err != nil {
+			continue
+		}
+		if jobTrigger, ok := workertypes.ToJobTrigger(triggerVal); ok {
+			jobTriggers = append(jobTriggers, jobTrigger)
+		}
+	}
+
+	return jobTriggers
+}

workers/push_delivery/pkg/dispatcher/dispatcher.go

@@ -159,7 +159,11 @@ func (g *deliveryJobGenerator) VisitV1(s workertypes.EventSummary) error {
 	// 2. Filter & Create Jobs
 	// Iterate Emails.
 	for _, sub := range subscribers.Emails {
-		if !shouldNotifyV1(sub.Triggers, s) {
+		notify, err := shouldNotifyV1(sub.Triggers, &s)
+		if err != nil {
+			return fmt.Errorf("error checking notification triggers for email subscription %s: %w", sub.SubscriptionID, err)
+		}
+		if !notify {
 			continue
 		}
 		g.emailJobs = append(g.emailJobs, workertypes.EmailDeliveryJob{
@@ -174,7 +178,11 @@ func (g *deliveryJobGenerator) VisitV1(s workertypes.EventSummary) error {
 
 	// Iterate Webhooks.
 	for _, sub := range subscribers.Webhooks {
-		if !shouldNotifyV1(sub.Triggers, s) {
+		notify, err := shouldNotifyV1(sub.Triggers, &s)
+		if err != nil {
+			return fmt.Errorf("error checking notification triggers for webhook subscription %s: %w", sub.SubscriptionID, err)
+		}
+		if !notify {
 			continue
 		}
 		g.webhookJobs = append(g.webhookJobs, workertypes.WebhookDeliveryJob{
@@ -196,40 +204,19 @@ func (g *deliveryJobGenerator) JobCount() int {
 	return len(g.emailJobs) + len(g.webhookJobs)
 }
 
-// shouldNotifyV1 determines if the V1 event summary matches any of the user's triggers.
-func shouldNotifyV1(triggers []workertypes.JobTrigger, summary workertypes.EventSummary) bool {
-	if len(summary.QueryErrors) > 0 || len(summary.ResolvedQueryErrors) > 0 {
-		return true
+// shouldNotifyV1 determines if the V1 event summary matches any of the subscriber's triggers.
+// Note: summary is passed as a pointer (*workertypes.EventSummary) to prevent copying the
+// EventSummary struct header and slice backing arrays across subscriber evaluation loops.
+// base.HasContent() returns true if there are active query errors, resolved query errors, or any
+// highlights matching the given triggers.
+func shouldNotifyV1(triggers []workertypes.JobTrigger, summary *workertypes.EventSummary) (bool, error) {
+	if summary == nil {
+		return false, nil
 	}
-
-	// 1. Determine if summary has changes.
-	hasChanges := summary.Categories.Added > 0 ||
-		summary.Categories.Removed > 0 ||
-		summary.Categories.Updated > 0 ||
-		summary.Categories.Moved > 0 ||
-		summary.Categories.Split > 0 ||
-		summary.Categories.QueryChanged > 0
-
-	if !hasChanges {
-		return false
-	}
-
-	// 2. Iterate triggers and check highlights.
-	for _, t := range triggers {
-		if matchesTrigger(t, summary) {
-			return true
-		}
-	}
-
-	return false
-}
-
-func matchesTrigger(t workertypes.JobTrigger, summary workertypes.EventSummary) bool {
-	for _, h := range summary.Highlights {
-		if h.MatchesTrigger(t) {
-			return true
-		}
+	base, err := summary.Categorize(triggers)
+	if err != nil {
+		return false, fmt.Errorf("failed to categorize event summary against triggers: %w", err)
 	}
 
-	return false
+	return base.HasContent(), nil
 }

workers/push_delivery/pkg/dispatcher/dispatcher_test.go

@@ -97,42 +97,21 @@ func createTestSummary(hasChanges bool) workertypes.EventSummary {
 		categories.Added = 1
 	}
 
-	return workertypes.EventSummary{
-		SchemaVersion:       "v1",
-		SnapshotOrigin:      workertypes.OriginLive,
-		Text:                "Test Summary",
-		Categories:          categories,
-		Truncated:           false,
-		QueryErrors:         nil,
-		ResolvedQueryErrors: nil,
-		Highlights:          nil,
-	}
+	summary := workertypes.NewEmptyEventSummary()
+	summary.SnapshotOrigin = workertypes.OriginLive
+	summary.Text = "Test Summary"
+	summary.Categories = categories
+
+	return summary
 }
 
 func createTestSummaryWithErrors(errCode workertypes.SummaryQueryErrorCode) workertypes.EventSummary {
-	categories := workertypes.SummaryCategories{
-		QueryChanged:    0,
-		Added:           0,
-		Deleted:         0,
-		Removed:         0,
-		Moved:           0,
-		Split:           0,
-		Updated:         0,
-		UpdatedImpl:     0,
-		UpdatedRename:   0,
-		UpdatedBaseline: 0,
-	}
+	summary := workertypes.NewEmptyEventSummary()
+	summary.SnapshotOrigin = workertypes.OriginLive
+	summary.Text = "Error occurred"
+	summary.SetQueryErrors([]workertypes.SummaryQueryError{{Code: errCode}})
 
-	return workertypes.EventSummary{
-		SchemaVersion:       workertypes.VersionEventSummaryV1,
-		SnapshotOrigin:      workertypes.OriginLive,
-		Text:                "Error occurred",
-		Categories:          categories,
-		Truncated:           false,
-		QueryErrors:         []workertypes.SummaryQueryError{{Code: errCode}},
-		ResolvedQueryErrors: nil,
-		Highlights:          nil,
-	}
+	return summary
 }
 
 // mockParserFactory creates a SummaryParser that injects the given summary directly.
@@ -205,22 +184,20 @@ func TestProcessEvent_Success(t *testing.T) {
 	summary := createTestSummary(true)
 	summary.Categories.UpdatedBaseline = 1
 	summary.Categories.Updated = 1
-	summary.Highlights = []workertypes.SummaryHighlight{
-		{
-			Type:        workertypes.SummaryHighlightTypeChanged,
-			FeatureID:   "test-feature-id",
-			FeatureName: "Test Feature",
-			Docs:        nil,
-			NameChange:  nil,
-			BaselineChange: &workertypes.Change[workertypes.BaselineValue]{
-				From: newBaselineValue(workertypes.BaselineStatusLimited),
-				To:   newBaselineValue(workertypes.BaselineStatusNewly),
-			},
-			BrowserChanges: nil,
-			Moved:          nil,
-			Split:          nil,
+	summary.AddHighlight(workertypes.SummaryHighlight{
+		Type:        workertypes.SummaryHighlightTypeChanged,
+		FeatureID:   "test-feature-id",
+		FeatureName: "Test Feature",
+		Docs:        nil,
+		NameChange:  nil,
+		BaselineChange: &workertypes.Change[workertypes.BaselineValue]{
+			From: newBaselineValue(workertypes.BaselineStatusLimited),
+			To:   newBaselineValue(workertypes.BaselineStatusNewly),
 		},
-	}
+		BrowserChanges: nil,
+		Moved:          nil,
+		Split:          nil,
+	})
 	parser := mockParserFactory(summary, nil)
 
 	d := NewDispatcher(finder, publisher)
@@ -315,22 +292,20 @@ func TestProcessEvent_Webhook_Success(t *testing.T) {
 	summary := createTestSummary(true)
 	summary.Categories.UpdatedBaseline = 1
 	summary.Categories.Updated = 1
-	summary.Highlights = []workertypes.SummaryHighlight{
-		{
-			Type:        workertypes.SummaryHighlightTypeChanged,
-			FeatureID:   "test-feature-id",
-			FeatureName: "Test Feature",
-			Docs:        nil,
-			NameChange:  nil,
-			BaselineChange: &workertypes.Change[workertypes.BaselineValue]{
-				From: newBaselineValue(workertypes.BaselineStatusLimited),
-				To:   newBaselineValue(workertypes.BaselineStatusNewly),
-			},
-			BrowserChanges: nil,
-			Moved:          nil,
-			Split:          nil,
+	summary.AddHighlight(workertypes.SummaryHighlight{
+		Type:        workertypes.SummaryHighlightTypeChanged,
+		FeatureID:   "test-feature-id",
+		FeatureName: "Test Feature",
+		Docs:        nil,
+		NameChange:  nil,
+		BaselineChange: &workertypes.Change[workertypes.BaselineValue]{
+			From: newBaselineValue(workertypes.BaselineStatusLimited),
+			To:   newBaselineValue(workertypes.BaselineStatusNewly),
 		},
-	}
+		BrowserChanges: nil,
+		Moved:          nil,
+		Split:          nil,
+	})
 	parser := mockParserFactory(summary, nil)
 
 	d := NewDispatcher(finder, publisher)
@@ -601,7 +576,7 @@ func newBrowserValue(status workertypes.BrowserStatus) workertypes.BrowserValue
 
 func withBaselineHighlight(
 	s workertypes.EventSummary, from, to workertypes.BaselineStatus) workertypes.EventSummary {
-	s.Highlights = append(s.Highlights, workertypes.SummaryHighlight{
+	s.AddHighlight(workertypes.SummaryHighlight{
 		Type:        workertypes.SummaryHighlightTypeChanged,
 		FeatureID:   "test-feature-id",
 		FeatureName: "Test Feature",
@@ -623,7 +598,7 @@ func withBaselineHighlight(
 
 func withBrowserChangeHighlight(
 	s workertypes.EventSummary, from, to workertypes.BrowserStatus) workertypes.EventSummary {
-	s.Highlights = append(s.Highlights, workertypes.SummaryHighlight{
+	s.AddHighlight(workertypes.SummaryHighlight{
 		Type:           workertypes.SummaryHighlightTypeChanged,
 		FeatureID:      "test-feature-id",
 		FeatureName:    "Test Feature",
@@ -764,7 +739,10 @@ func TestShouldNotifyV1(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			got := shouldNotifyV1(tc.triggers, tc.summary)
+			got, err := shouldNotifyV1(tc.triggers, &tc.summary)
+			if err != nil {
+				t.Fatalf("shouldNotifyV1 unexpected error: %v", err)
+			}
 			if got != tc.want {
 				t.Errorf("shouldNotifyV1() = %v, want %v", got, tc.want)
 			}
@@ -773,16 +751,12 @@ func TestShouldNotifyV1(t *testing.T) {
 }
 
 func createRecoveredSummary() workertypes.EventSummary {
-	return workertypes.EventSummary{
-		SchemaVersion:       workertypes.VersionEventSummaryV1,
-		SnapshotOrigin:      workertypes.OriginLive,
-		Text:                "Search query recovered and tracking 2 features normally.",
-		Truncated:           false,
-		Categories:          workertypes.NewEmptySummaryCategories(),
-		QueryErrors:         nil,
-		ResolvedQueryErrors: []workertypes.SummaryQueryError{{Code: workertypes.SummaryQueryErrorCodeQueryGrammar}},
-		Highlights:          nil,
-	}
+	summary := workertypes.NewEmptyEventSummary()
+	summary.SnapshotOrigin = workertypes.OriginLive
+	summary.Text = "Search query recovered and tracking 2 features normally."
+	summary.SetResolvedQueryErrors([]workertypes.SummaryQueryError{{Code: workertypes.SummaryQueryErrorCodeQueryGrammar}})
+
+	return summary
 }
 
 func TestShouldNotifyV1_ResolvedQueryErrors(t *testing.T) {
@@ -816,7 +790,10 @@ func TestShouldNotifyV1_ResolvedQueryErrors(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			got := shouldNotifyV1(tc.triggers, tc.summary)
+			got, err := shouldNotifyV1(tc.triggers, &tc.summary)
+			if err != nil {
+				t.Fatalf("shouldNotifyV1 unexpected error: %v", err)
+			}
 			if got != tc.want {
 				t.Errorf("shouldNotifyV1() = %v, want %v", got, tc.want)
 			}
@@ -878,3 +855,14 @@ func TestProcessEvent_ResolvedQueryErrors_AllChannels(t *testing.T) {
 		t.Errorf("expected subscription ID sub-webhook-recovery, got %s", publisher.webhookJobs[0].SubscriptionID)
 	}
 }
+
+func TestShouldNotifyV1_NilSummary(t *testing.T) {
+	// Verify that if the summary is nil, shouldNotifyV1 returns false and no error.
+	got, err := shouldNotifyV1([]workertypes.JobTrigger{workertypes.FeaturePromotedToNewly}, nil)
+	if err != nil {
+		t.Fatalf("shouldNotifyV1 unexpected error: %v", err)
+	}
+	if got {
+		t.Errorf("shouldNotifyV1(triggers, nil) = %v, want false", got)
+	}
+}