Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jcscottiii wants to merge 1 commit into
base: feat/all-saved-search-slack-visitor-impl
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 29 additions & 14 deletions backend/pkg/httpserver/get_subscription_rss.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ package httpserver
import (
"bytes"
"context"
"encoding/json"
"encoding/xml"
"errors"
"fmt"
Expand Down Expand Up @@ -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(
Expand All @@ -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
}
Expand Down Expand Up @@ -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
}
57 changes: 22 additions & 35 deletions workers/push_delivery/pkg/dispatcher/dispatcher.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand All @@ -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{
Expand All @@ -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
}
139 changes: 63 additions & 76 deletions workers/push_delivery/pkg/dispatcher/dispatcher_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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",
Expand All @@ -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",
Expand Down Expand Up @@ -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)
}
Expand All @@ -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) {
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -878,3 +855,13 @@ 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) {
got, err := shouldNotifyV1([]workertypes.JobTrigger{workertypes.FeaturePromotedToNewly}, nil)
if err != nil {
t.Fatalf("shouldNotifyV1 unexpected error: %v", err)
}
if got != false {
t.Errorf("shouldNotifyV1(triggers, nil) = %v, want false", got)
}
}