From 5f26bf4b158981cb46a0d78bc689b30ad240c827 Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Fri, 26 Jun 2026 09:19:29 +0700 Subject: [PATCH] refactor(poller): batch delivery checks and dedupe component helpers Replace the per-subscriber SISMEMBER delivery check with a single SMEMBERS per event plus in-memory lookups, cutting N round-trips to one. Remove the component-label helpers duplicated between poller and bot by exporting the canonical versions from poller; bot already depends on it. Collapse the repeated component checkpoint closures into a shared helper. Drop the redundant escapeAttr alias and the unused variadic on bot.New. --- internal/bot/bot.go | 8 ++---- internal/bot/commands.go | 3 ++- internal/bot/component_resolution.go | 38 ++++------------------------ internal/bot/format.go | 20 ++++++--------- internal/bot/helpers.go | 3 ++- internal/poller/delivery.go | 18 +++++++------ internal/poller/event_collection.go | 33 ++++++++++-------------- internal/poller/format.go | 14 ++++++---- internal/poller/poller.go | 2 +- internal/poller/poller_test.go | 8 +++--- internal/redisstore/checkpoint.go | 15 +++++++++-- 11 files changed, 70 insertions(+), 92 deletions(-) diff --git a/internal/bot/bot.go b/internal/bot/bot.go index 4a75a25..b8526c9 100644 --- a/internal/bot/bot.go +++ b/internal/bot/bot.go @@ -40,17 +40,13 @@ type Bot struct { username string } -func New(telegramClient TelegramClient, statusClient StatusClient, store Store, logger *slog.Logger, username ...string) *Bot { - botUsername := "" - if len(username) > 0 { - botUsername = username[0] - } +func New(telegramClient TelegramClient, statusClient StatusClient, store Store, logger *slog.Logger, username string) *Bot { return &Bot{ telegramClient: telegramClient, statusClient: statusClient, store: store, logger: logger, - username: botUsername, + username: username, } } diff --git a/internal/bot/commands.go b/internal/bot/commands.go index f8de588..ca6bcfe 100644 --- a/internal/bot/commands.go +++ b/internal/bot/commands.go @@ -5,6 +5,7 @@ import ( "fmt" "strings" + "github.com/tiennm99/openai-status-bot/internal/poller" "github.com/tiennm99/openai-status-bot/internal/redisstore" "github.com/tiennm99/openai-status-bot/internal/telegram" ) @@ -128,7 +129,7 @@ func (b *Bot) updateComponentFilter(ctx context.Context, message telegram.Messag b.reply(ctx, message, "Not subscribed yet. Use /start first.") return } - b.reply(ctx, message, fmt.Sprintf("Subscribed to component: %s\nActive filters: %s", escape(componentLabel(component, duplicateComponentNames(summary.Components)[component.Name])), escape(componentFilterLabels(summary.Components, components)))) + b.reply(ctx, message, fmt.Sprintf("Subscribed to component: %s\nActive filters: %s", escape(poller.ComponentLabel(component, poller.DuplicateComponentNames(summary.Components)[component.Name])), escape(componentFilterLabels(summary.Components, components)))) } func withSubscriptionType(types []string, subscriptionType string) []string { diff --git a/internal/bot/component_resolution.go b/internal/bot/component_resolution.go index 46993e2..6551cda 100644 --- a/internal/bot/component_resolution.go +++ b/internal/bot/component_resolution.go @@ -5,6 +5,7 @@ import ( "strings" openai "github.com/tiennm99/openai-status-bot/internal/openai" + "github.com/tiennm99/openai-status-bot/internal/poller" ) type componentResolution struct { @@ -64,10 +65,10 @@ func matchingComponents(components []openai.Component, query string, matches fun } func formatAmbiguousComponents(query string, matches []openai.Component) string { - duplicates := duplicateComponentNames(matches) + duplicates := poller.DuplicateComponentNames(matches) lines := []string{fmt.Sprintf("Component %s is ambiguous. Use one of these IDs:", escape(query)), ""} for _, component := range matches { - lines = append(lines, fmt.Sprintf("- %s: %s", escape(componentLabel(component, duplicates[component.Name])), escape(component.ID))) + lines = append(lines, fmt.Sprintf("- %s: %s", escape(poller.ComponentLabel(component, duplicates[component.Name])), escape(component.ID))) } return truncateMessage(strings.Join(lines, "\n")) } @@ -76,13 +77,13 @@ func componentFilterLabels(components []openai.Component, ids []string) string { if len(ids) == 0 { return "all" } - duplicates := duplicateComponentNames(components) + duplicates := poller.DuplicateComponentNames(components) labels := make([]string, 0, len(ids)) for _, id := range ids { label := id for _, component := range components { if component.ID == id { - label = componentLabel(component, duplicates[component.Name]) + label = poller.ComponentLabel(component, duplicates[component.Name]) break } } @@ -90,32 +91,3 @@ func componentFilterLabels(components []openai.Component, ids []string) string { } return strings.Join(labels, ", ") } - -func duplicateComponentNames(components []openai.Component) map[string]bool { - counts := map[string]int{} - for _, component := range components { - if component.Group { - continue - } - counts[component.Name]++ - } - duplicates := map[string]bool{} - for name, count := range counts { - duplicates[name] = count > 1 - } - return duplicates -} - -func componentLabel(component openai.Component, duplicate bool) string { - if !duplicate || component.ID == "" { - return component.Name - } - return fmt.Sprintf("%s (ID: %s)", component.Name, shortID(component.ID)) -} - -func shortID(value string) string { - if len(value) <= 8 { - return value - } - return value[:8] -} diff --git a/internal/bot/format.go b/internal/bot/format.go index 2e9bb68..4dcdd86 100644 --- a/internal/bot/format.go +++ b/internal/bot/format.go @@ -33,13 +33,13 @@ func formatStatus(summary openai.Summary) string { fmt.Sprintf("Overall: %s", escape(summary.Status.Description)), } - duplicates := duplicateComponentNames(summary.Components) + duplicates := poller.DuplicateComponentNames(summary.Components) degraded := make([]string, 0) for _, component := range summary.Components { if component.Group || component.Status == "operational" { continue } - degraded = append(degraded, fmt.Sprintf("- %s: %s", escape(componentLabel(component, duplicates[component.Name])), escape(poller.StatusLabel(component.Status)))) + degraded = append(degraded, fmt.Sprintf("- %s: %s", escape(poller.ComponentLabel(component, duplicates[component.Name])), escape(poller.StatusLabel(component.Status)))) } if len(degraded) == 0 { @@ -56,7 +56,7 @@ func formatStatus(summary openai.Summary) string { func formatComponentStatus(component openai.Component, duplicate bool) string { return fmt.Sprintf( "%s\n\nStatus: %s\nLast change: %s UTC\n\nView full status page", - escape(componentLabel(component, duplicate)), + escape(poller.ComponentLabel(component, duplicate)), escape(poller.StatusLabel(component.Status)), escape(formatTime(component.UpdatedAt)), statusURL, @@ -65,12 +65,12 @@ func formatComponentStatus(component openai.Component, duplicate bool) string { func formatComponents(summary openai.Summary) string { lines := []string{"OpenAI Components", ""} - duplicates := duplicateComponentNames(summary.Components) + duplicates := poller.DuplicateComponentNames(summary.Components) for _, component := range summary.Components { if component.Group { continue } - lines = append(lines, fmt.Sprintf("- %s: %s", escape(componentLabel(component, duplicates[component.Name])), escape(poller.StatusLabel(component.Status)))) + lines = append(lines, fmt.Sprintf("- %s: %s", escape(poller.ComponentLabel(component, duplicates[component.Name])), escape(poller.StatusLabel(component.Status)))) } return truncateMessage(strings.Join(lines, "\n")) } @@ -99,7 +99,7 @@ func formatHistory(incidents []openai.Incident, count int) string { escape(poller.StatusLabel(incident.Status)), ) if link != "" { - entry += fmt.Sprintf("\n Details", escapeAttr(link)) + entry += fmt.Sprintf("\n Details", escape(link)) } lines = append(lines, entry) } @@ -109,14 +109,14 @@ func formatHistory(incidents []openai.Incident, count int) string { func formatUptime(summary openai.Summary) string { lines := []string{"OpenAI Component Health", ""} - duplicates := duplicateComponentNames(summary.Components) + duplicates := poller.DuplicateComponentNames(summary.Components) for _, component := range summary.Components { if component.Group { continue } lines = append(lines, fmt.Sprintf( "%s\n Status: %s\n Last change: %s UTC", - escape(componentLabel(component, duplicates[component.Name])), + escape(poller.ComponentLabel(component, duplicates[component.Name])), escape(poller.StatusLabel(component.Status)), escape(formatTime(component.UpdatedAt)), )) @@ -179,7 +179,3 @@ func emptyDefault(value, fallback string) string { func escape(value string) string { return html.EscapeString(value) } - -func escapeAttr(value string) string { - return html.EscapeString(value) -} diff --git a/internal/bot/helpers.go b/internal/bot/helpers.go index e175555..68699a9 100644 --- a/internal/bot/helpers.go +++ b/internal/bot/helpers.go @@ -5,6 +5,7 @@ import ( "fmt" "strings" + "github.com/tiennm99/openai-status-bot/internal/poller" "github.com/tiennm99/openai-status-bot/internal/redisstore" "github.com/tiennm99/openai-status-bot/internal/telegram" ) @@ -27,7 +28,7 @@ func (b *Bot) replyStatus(ctx context.Context, message telegram.Message, fields b.reply(ctx, message, formatAmbiguousComponents(query, resolution.Matches)) return } - duplicates := duplicateComponentNames(summary.Components) + duplicates := poller.DuplicateComponentNames(summary.Components) b.reply(ctx, message, formatComponentStatus(resolution.Component, duplicates[resolution.Component.Name])) return } diff --git a/internal/poller/delivery.go b/internal/poller/delivery.go index 4d64dcc..43d774e 100644 --- a/internal/poller/delivery.go +++ b/internal/poller/delivery.go @@ -46,6 +46,14 @@ func (e *deliveryError) addAll(other *deliveryError) { func (r *Runner) notifySubscribers(ctx context.Context, event notificationEvent, subscribers []redisstore.Subscriber, removed, failed map[string]bool) (*deliveryError, error) { deliveryFailures := &deliveryError{} + var delivered map[string]bool + if event.deliveryKey != "" { + var err error + delivered, err = r.store.DeliveredSubscribers(ctx, event.deliveryKey) + if err != nil { + return nil, err + } + } for _, subscriber := range subscribers { subscriberKey := subscriber.Key() if removed[subscriberKey] || failed[subscriberKey] { @@ -54,14 +62,8 @@ func (r *Runner) notifySubscribers(ctx context.Context, event notificationEvent, if !subscriber.Accepts(event.eventType, event.componentID, event.componentName) { continue } - if event.deliveryKey != "" { - delivered, err := r.store.HasDelivered(ctx, event.deliveryKey, subscriberKey) - if err != nil { - return nil, err - } - if delivered { - continue - } + if delivered[subscriberKey] { + continue } if err := r.notifier.SendMessage(ctx, subscriber, event.text); err != nil { if telegram.IsTerminalSendError(err) { diff --git a/internal/poller/event_collection.go b/internal/poller/event_collection.go index 20c5f44..9420f3e 100644 --- a/internal/poller/event_collection.go +++ b/internal/poller/event_collection.go @@ -45,7 +45,7 @@ func (r *Runner) collectComponentEvents(ctx context.Context, summary openai.Summ return nil, nil, nil, err } - duplicates := duplicateComponentNames(componentsForDuplicateLabels(summary.Components, pending)) + duplicates := DuplicateComponentNames(componentsForDuplicateLabels(summary.Components, pending)) events := make([]notificationEvent, 0) before := make([]checkpoint, 0) @@ -62,15 +62,7 @@ func (r *Runner) collectComponentEvents(ctx context.Context, summary openai.Summ sortTime: pendingEvent.UpdatedAt, text: FormatComponentChange(component, pendingEvent.PreviousStatus, duplicates[component.Name]), }) - after = append(after, func(ctx context.Context) error { - return r.store.SaveComponentStatus(ctx, pendingEvent.ComponentID, pendingEvent.Status) - }) - after = append(after, func(ctx context.Context) error { - return r.store.RemovePendingComponentEvent(ctx, pendingEvent.ComponentID) - }) - after = append(after, func(ctx context.Context) error { - return r.store.ClearDelivery(ctx, pendingEvent.DeliveryKey) - }) + after = append(after, r.resolveComponentCheckpoints(pendingEvent.ComponentID, pendingEvent.Status, pendingEvent.DeliveryKey)...) } for _, component := range summary.Components { @@ -120,19 +112,22 @@ func (r *Runner) collectComponentEvents(ctx context.Context, summary openai.Summ sortTime: component.UpdatedAt, text: FormatComponentChange(component, previousStatus, duplicates[component.Name]), }) - after = append(after, func(ctx context.Context) error { - return r.store.SaveComponentStatus(ctx, component.ID, component.Status) - }) - after = append(after, func(ctx context.Context) error { - return r.store.RemovePendingComponentEvent(ctx, component.ID) - }) - after = append(after, func(ctx context.Context) error { - return r.store.ClearDelivery(ctx, deliveryKey) - }) + after = append(after, r.resolveComponentCheckpoints(component.ID, component.Status, deliveryKey)...) } return events, before, after, nil } +// resolveComponentCheckpoints builds the post-delivery writes shared by pending +// and freshly detected component events: persist the new status, drop the +// pending marker, and clear the delivery-state set. +func (r *Runner) resolveComponentCheckpoints(componentID, status, deliveryKey string) []checkpoint { + return []checkpoint{ + func(ctx context.Context) error { return r.store.SaveComponentStatus(ctx, componentID, status) }, + func(ctx context.Context) error { return r.store.RemovePendingComponentEvent(ctx, componentID) }, + func(ctx context.Context) error { return r.store.ClearDelivery(ctx, deliveryKey) }, + } +} + func (r *Runner) collectIncidentEvents(ctx context.Context, response openai.IncidentsResponse, initialized bool) ([]notificationEvent, []checkpoint, error) { events := make([]notificationEvent, 0) checkpoints := make([]checkpoint, 0) diff --git a/internal/poller/format.go b/internal/poller/format.go index 915a32c..a4a4d9c 100644 --- a/internal/poller/format.go +++ b/internal/poller/format.go @@ -11,7 +11,7 @@ import ( func FormatComponentChange(component openai.Component, previousStatus string, duplicate bool) string { return fmt.Sprintf( "OpenAI component update\n\nComponent: %s\nStatus: %s -> %s\n\nView full status page", - escape(componentLabel(component, duplicate)), + escape(ComponentLabel(component, duplicate)), escape(StatusLabel(previousStatus)), escape(StatusLabel(component.Status)), ) @@ -75,7 +75,9 @@ func StatusLabel(value string) string { } } -func duplicateComponentNames(components []openai.Component) map[string]bool { +// DuplicateComponentNames reports which non-group component names appear more +// than once, so labels can disambiguate them with a short ID suffix. +func DuplicateComponentNames(components []openai.Component) map[string]bool { counts := map[string]int{} for _, component := range components { if component.Group { @@ -90,14 +92,16 @@ func duplicateComponentNames(components []openai.Component) map[string]bool { return duplicates } -func componentLabel(component openai.Component, duplicate bool) string { +// ComponentLabel renders a component's display name, appending a short ID when +// the name collides with another component. +func ComponentLabel(component openai.Component, duplicate bool) string { if !duplicate || component.ID == "" { return component.Name } - return fmt.Sprintf("%s (ID: %s)", component.Name, shortID(component.ID)) + return fmt.Sprintf("%s (ID: %s)", component.Name, ShortID(component.ID)) } -func shortID(value string) string { +func ShortID(value string) string { if len(value) <= 8 { return value } diff --git a/internal/poller/poller.go b/internal/poller/poller.go index 5f3445f..31b1888 100644 --- a/internal/poller/poller.go +++ b/internal/poller/poller.go @@ -17,7 +17,7 @@ type StatusClient interface { type Store interface { ComponentStatuses(ctx context.Context) (map[string]string, error) ClearDelivery(ctx context.Context, eventKey string) error - HasDelivered(ctx context.Context, eventKey, subscriberKey string) (bool, error) + DeliveredSubscribers(ctx context.Context, eventKey string) (map[string]bool, error) HasIncidentUpdateVersion(ctx context.Context, updateID, version string) (bool, error) IsInitialized(ctx context.Context) (bool, error) ListSubscribers(ctx context.Context) ([]redisstore.Subscriber, error) diff --git a/internal/poller/poller_test.go b/internal/poller/poller_test.go index 3dfb0b2..52957bb 100644 --- a/internal/poller/poller_test.go +++ b/internal/poller/poller_test.go @@ -61,11 +61,11 @@ func (f *fakePollerStore) ClearDelivery(_ context.Context, eventKey string) erro return nil } -func (f *fakePollerStore) HasDelivered(_ context.Context, eventKey, subscriberKey string) (bool, error) { +func (f *fakePollerStore) DeliveredSubscribers(_ context.Context, eventKey string) (map[string]bool, error) { if f.hasDeliveredErr != nil { - return false, f.hasDeliveredErr + return nil, f.hasDeliveredErr } - return f.delivered[eventKey][subscriberKey], nil + return f.delivered[eventKey], nil } func (f *fakePollerStore) HasIncidentUpdateVersion(_ context.Context, updateID, version string) (bool, error) { @@ -407,7 +407,7 @@ func TestPendingComponentDuplicateLabelsIncludeCurrentRename(t *testing.T) { {ID: "c2", Name: "API", Status: "operational"}, } - duplicates := duplicateComponentNames(componentsForDuplicateLabels(components, pending)) + duplicates := DuplicateComponentNames(componentsForDuplicateLabels(components, pending)) if !duplicates["API"] { t.Fatalf("duplicates = %v, want current API rename counted as duplicate", duplicates) } diff --git a/internal/redisstore/checkpoint.go b/internal/redisstore/checkpoint.go index 82c80fb..3f7e1dd 100644 --- a/internal/redisstore/checkpoint.go +++ b/internal/redisstore/checkpoint.go @@ -115,8 +115,19 @@ func (s *Store) MarkIncidentUpdateVersion(ctx context.Context, updateID, version return err } -func (s *Store) HasDelivered(ctx context.Context, eventKey, subscriberKey string) (bool, error) { - return s.client.SIsMember(ctx, deliveryStateKey(eventKey), subscriberKey).Result() +// DeliveredSubscribers returns the set of subscriber keys already notified for +// eventKey. Callers fan a single event out to every subscriber, so one SMEMBERS +// plus in-memory lookups replaces one SISMEMBER per subscriber. +func (s *Store) DeliveredSubscribers(ctx context.Context, eventKey string) (map[string]bool, error) { + members, err := s.client.SMembers(ctx, deliveryStateKey(eventKey)).Result() + if err != nil { + return nil, err + } + delivered := make(map[string]bool, len(members)) + for _, member := range members { + delivered[member] = true + } + return delivered, nil } func (s *Store) MarkDelivered(ctx context.Context, eventKey, subscriberKey string) error {