diff --git a/README.md b/README.md index 18cac83..232d57e 100644 --- a/README.md +++ b/README.md @@ -7,8 +7,10 @@ Telegram bot that watches [OpenAI Status](https://status.openai.com/) every minu - Checks OpenAI status on a configurable interval, default `1m` - Notifies subscribers about new incident updates - Notifies subscribers when component status changes -- Uses Redis for subscribers, component status checkpoints, and seen incident updates +- Supports incident-only, component-only, and component-filtered subscriptions +- Uses Redis for subscribers, delivery state, component checkpoints, and seen incident update versions - Supports Telegram supergroup topics via `message_thread_id` +- Clears an existing Telegram webhook before long polling, for migration from webhook deployments - Includes Docker Compose for local Redis + bot runtime ## Bot Commands @@ -17,9 +19,13 @@ Telegram bot that watches [OpenAI Status](https://status.openai.com/) every minu |---------|-------------| | `/start` | Subscribe current chat or topic | | `/stop` | Unsubscribe current chat or topic | -| `/status` | Show current OpenAI status | +| `/status [component]` | Show current OpenAI status, optionally for one component | | `/components` | Show all OpenAI component statuses | +| `/subscribe ` | Set notification types | +| `/subscribe component ` | Filter component notifications or clear the filter | | `/history [count]` | Show recent incidents, default 5, max 10 | +| `/uptime` | Show component health overview | +| `/info` | Show chat ID and subscription settings | | `/help` | Show command help | ## Quick Start @@ -53,3 +59,5 @@ go run ./cmd/openai-status-bot ## Notes The first successful poll seeds Redis and does not send historical incidents. Notifications start from later changes. + +Incident update dedupe tracks the update content/version, so edited Statuspage updates can notify again. Delivery is checkpointed after successful fan-out; retryable Telegram failures may be retried on a later poll without advancing the global checkpoint. diff --git a/cmd/openai-status-bot/main.go b/cmd/openai-status-bot/main.go index bfa06ec..f822880 100644 --- a/cmd/openai-status-bot/main.go +++ b/cmd/openai-status-bot/main.go @@ -45,9 +45,20 @@ func main() { store := redisstore.New(redisClient) statusClient := openai.NewClient(cfg.OpenAIStatusBaseURL, cfg.HTTPTimeout) telegramClient := telegram.NewClient(cfg.TelegramBotToken, cfg.HTTPTimeout) + if err := telegramClient.DeleteWebhook(ctx); err != nil { + logger.Error("delete telegram webhook", "error", err) + os.Exit(1) + } + + botUsername := "" + if me, err := telegramClient.GetMe(ctx); err != nil { + logger.Warn("get telegram bot profile", "error", err) + } else { + botUsername = me.Username + } statusPoller := poller.NewRunner(statusClient, store, telegramClient, cfg.PollInterval, logger) - commandBot := bot.New(telegramClient, statusClient, store, logger) + commandBot := bot.New(telegramClient, statusClient, store, logger, botUsername) go statusPoller.Run(ctx) diff --git a/docs/setup-guide.md b/docs/setup-guide.md index 5480a42..e1060c7 100644 --- a/docs/setup-guide.md +++ b/docs/setup-guide.md @@ -37,4 +37,4 @@ go test ./... go build ./cmd/openai-status-bot ``` -Then send `/start` to the Telegram bot and run `/status`. +Then send `/start` to the Telegram bot and run `/status`. If the same token was previously used by a webhook deployment, startup clears that webhook before long polling begins. diff --git a/docs/system-architecture.md b/docs/system-architecture.md index 3b1694c..58cba2c 100644 --- a/docs/system-architecture.md +++ b/docs/system-architecture.md @@ -7,36 +7,45 @@ - Telegram long polling loop for user commands. - OpenAI status polling loop, default every minute. -Redis stores subscribers and polling checkpoints. +Redis stores subscribers, subscription settings, polling checkpoints, delivery retry state, and the Telegram update offset. ## Data Flow 1. User sends `/start` in Telegram. -2. Bot stores chat ID and optional topic thread ID in Redis. -3. Poller fetches OpenAI status JSON: +2. Bot stores chat ID, optional topic thread ID, and default subscription settings in Redis. +3. Users can adjust settings with `/subscribe` and inspect them with `/info`. +4. Poller fetches OpenAI status JSON: - `GET /api/v2/summary.json` - `GET /api/v2/incidents.json` -4. Poller compares fetched state with Redis checkpoints. -5. New incident updates or component status changes are sent to all subscribers. +5. Poller compares fetched state with Redis checkpoints and builds notification events without mutating checkpoints. +6. Events are sent to eligible subscribers, respecting incident/component preferences and component ID filters. +7. Component and incident checkpoints are written only after delivery succeeds or terminal subscriber failures are removed. ## Redis Keys | Key | Type | Purpose | |-----|------|---------| | `openai-status:subscribers` | set | Telegram chat or topic subscribers | +| `openai-status:subscriber-settings` | hash | Subscription types and component ID filters by subscriber key | | `openai-status:component-statuses` | hash | Last seen component status by component ID | -| `openai-status:incident-updates` | set | Seen incident update IDs | +| `openai-status:incident-updates` | set | Legacy seen incident update IDs, retained for migration | +| `openai-status:incident-update-versions` | hash | Seen incident update version by update ID | +| `openai-status:event-delivery:` | set | Temporary per-event subscriber delivery state for retry isolation | +| `openai-status:telegram-offset` | string | Last processed Telegram update offset | | `openai-status:initialized` | string | Baseline seed marker | -Subscriber set members are `chatID` or `chatID:threadID`. +Subscriber set members are `chatID` or `chatID:threadID`. Missing settings default to incident and component notifications for backward compatibility with older Redis state. ## Runtime -The service uses Telegram `getUpdates`, so it does not need a public webhook URL. Docker Compose starts Redis and the bot. +The service uses Telegram `getUpdates`, so it does not need a public webhook URL. On startup it calls `deleteWebhook` before long polling, which allows migration from a webhook-based deployment using the same bot token. Docker Compose starts Redis and the bot. ## Failure Behavior - First successful poll seeds state without notification. - OpenAI fetch failures are logged and retried on the next interval. -- Telegram send failures are logged; other subscribers still receive messages. +- Retryable Telegram failures do not advance component or incident checkpoints. +- Successful per-subscriber deliveries are tracked temporarily, so retrying one failed subscriber does not resend to already-delivered subscribers. +- Telegram 403 and selected terminal 400 errors remove the unreachable subscriber, then delivery continues. +- Malformed subscriber keys are skipped instead of blocking all fan-out. - Redis connection failure at startup exits the process. diff --git a/internal/bot/bot.go b/internal/bot/bot.go index 1799237..134b3b1 100644 --- a/internal/bot/bot.go +++ b/internal/bot/bot.go @@ -23,7 +23,12 @@ type StatusClient interface { type Store interface { AddSubscriber(ctx context.Context, sub redisstore.Subscriber) error + GetSubscriber(ctx context.Context, sub redisstore.Subscriber) (redisstore.Subscriber, bool, error) RemoveSubscriber(ctx context.Context, sub redisstore.Subscriber) error + SaveTelegramOffset(ctx context.Context, offset int64) error + TelegramOffset(ctx context.Context) (int64, error) + UpdateSubscriberComponents(ctx context.Context, sub redisstore.Subscriber, components []string) (bool, error) + UpdateSubscriberTypes(ctx context.Context, sub redisstore.Subscriber, types []string) (bool, error) } type Bot struct { @@ -31,19 +36,28 @@ type Bot struct { statusClient StatusClient store Store logger *slog.Logger + username string } -func New(telegramClient TelegramClient, statusClient StatusClient, store Store, logger *slog.Logger) *Bot { +func New(telegramClient TelegramClient, statusClient StatusClient, store Store, logger *slog.Logger, username ...string) *Bot { + botUsername := "" + if len(username) > 0 { + botUsername = username[0] + } return &Bot{ telegramClient: telegramClient, statusClient: statusClient, store: store, logger: logger, + username: botUsername, } } func (b *Bot) Run(ctx context.Context) error { - var offset int64 + offset, err := b.store.TelegramOffset(ctx) + if err != nil { + b.logger.Warn("load telegram offset", "error", err) + } for { select { @@ -63,11 +77,16 @@ func (b *Bot) Run(ctx context.Context) error { } for _, update := range updates { - offset = update.UpdateID + 1 - if update.Message == nil { + if update.UpdateID < offset { continue } - b.handleMessage(ctx, *update.Message) + if update.Message != nil { + b.handleMessage(ctx, *update.Message) + } + offset = update.UpdateID + 1 + if err := b.store.SaveTelegramOffset(ctx, offset); err != nil { + b.logger.Warn("save telegram offset", "offset", offset, "error", err) + } } } } @@ -77,18 +96,26 @@ func (b *Bot) handleMessage(ctx context.Context, message telegram.Message) { return } - command, fields := normalizeCommand(message.Text) + command, fields := normalizeCommand(message.Text, b.username) switch command { + case "": + return case "/start": b.subscribe(ctx, message) case "/stop": b.unsubscribe(ctx, message) case "/status": - b.replyStatus(ctx, message) + b.replyStatus(ctx, message, fields) case "/components": b.replyComponents(ctx, message) + case "/subscribe": + b.replySubscribe(ctx, message, fields) case "/history": b.replyHistory(ctx, message, parseHistoryCount(fields)) + case "/uptime": + b.replyUptime(ctx, message) + case "/info": + b.replyInfo(ctx, message) case "/help": b.reply(ctx, message, helpText) default: @@ -103,7 +130,7 @@ func (b *Bot) subscribe(ctx context.Context, message telegram.Message) { b.reply(ctx, message, "Could not subscribe right now.") return } - b.reply(ctx, message, "Subscribed to OpenAI status updates.") + b.reply(ctx, message, "Subscribed to OpenAI status updates (incidents + components). Use /subscribe to change preferences.") } func (b *Bot) unsubscribe(ctx context.Context, message telegram.Message) { @@ -113,41 +140,5 @@ func (b *Bot) unsubscribe(ctx context.Context, message telegram.Message) { b.reply(ctx, message, "Could not unsubscribe right now.") return } - b.reply(ctx, message, "Unsubscribed from OpenAI status updates.") -} - -func (b *Bot) replyStatus(ctx context.Context, message telegram.Message) { - summary, err := b.statusClient.FetchSummary(ctx) - if err != nil { - b.logger.Error("fetch status", "error", err) - b.reply(ctx, message, "Could not fetch OpenAI status right now.") - return - } - b.reply(ctx, message, formatStatus(summary)) -} - -func (b *Bot) replyComponents(ctx context.Context, message telegram.Message) { - summary, err := b.statusClient.FetchSummary(ctx) - if err != nil { - b.logger.Error("fetch components", "error", err) - b.reply(ctx, message, "Could not fetch OpenAI components right now.") - return - } - b.reply(ctx, message, formatComponents(summary)) -} - -func (b *Bot) replyHistory(ctx context.Context, message telegram.Message, count int) { - incidents, err := b.statusClient.FetchIncidents(ctx) - if err != nil { - b.logger.Error("fetch incidents", "error", err) - b.reply(ctx, message, "Could not fetch OpenAI incident history right now.") - return - } - b.reply(ctx, message, formatHistory(incidents.Incidents, count)) -} - -func (b *Bot) reply(ctx context.Context, message telegram.Message, text string) { - if err := b.telegramClient.SendText(ctx, message.Chat.ID, message.MessageThreadID, text); err != nil { - b.logger.Warn("send telegram reply", "chat_id", message.Chat.ID, "error", err) - } + b.reply(ctx, message, "Unsubscribed from OpenAI status updates. Use /start to resubscribe.") } diff --git a/internal/bot/bot_test.go b/internal/bot/bot_test.go new file mode 100644 index 0000000..674e1ee --- /dev/null +++ b/internal/bot/bot_test.go @@ -0,0 +1,155 @@ +package bot + +import ( + "context" + "log/slog" + "strings" + "testing" + + openai "github.com/tiennm99/openai-status-bot/internal/openai" + "github.com/tiennm99/openai-status-bot/internal/redisstore" + "github.com/tiennm99/openai-status-bot/internal/telegram" +) + +type fakeTelegramClient struct { + replies []string +} + +func (f *fakeTelegramClient) GetUpdates(context.Context, int64, int) ([]telegram.Update, error) { + return nil, nil +} + +func (f *fakeTelegramClient) SendText(_ context.Context, _ int64, _ *int, text string) error { + f.replies = append(f.replies, text) + return nil +} + +type fakeBotStatusClient struct { + summary openai.Summary +} + +func (f fakeBotStatusClient) FetchSummary(context.Context) (openai.Summary, error) { + return f.summary, nil +} + +func (f fakeBotStatusClient) FetchIncidents(context.Context) (openai.IncidentsResponse, error) { + return openai.IncidentsResponse{}, nil +} + +type fakeBotStore struct { + sub redisstore.Subscriber + subscribed bool + types []string + components []string +} + +func (f *fakeBotStore) AddSubscriber(_ context.Context, sub redisstore.Subscriber) error { + f.sub = sub + f.subscribed = true + f.types = redisstore.DefaultSubscriptionTypes() + return nil +} + +func (f *fakeBotStore) GetSubscriber(context.Context, redisstore.Subscriber) (redisstore.Subscriber, bool, error) { + if !f.subscribed { + return redisstore.Subscriber{}, false, nil + } + sub := f.sub + sub.Types = append([]string{}, f.types...) + sub.Components = append([]string{}, f.components...) + return sub, true, nil +} + +func (f *fakeBotStore) RemoveSubscriber(context.Context, redisstore.Subscriber) error { + f.subscribed = false + return nil +} + +func (f *fakeBotStore) SaveTelegramOffset(context.Context, int64) error { return nil } +func (f *fakeBotStore) TelegramOffset(context.Context) (int64, error) { return 0, nil } + +func (f *fakeBotStore) UpdateSubscriberComponents(_ context.Context, _ redisstore.Subscriber, components []string) (bool, error) { + if !f.subscribed { + return false, nil + } + f.components = append([]string{}, components...) + return true, nil +} + +func (f *fakeBotStore) UpdateSubscriberTypes(_ context.Context, _ redisstore.Subscriber, types []string) (bool, error) { + if !f.subscribed { + return false, nil + } + f.types = append([]string{}, types...) + return true, nil +} + +func TestSubscribeComponentStoresComponentID(t *testing.T) { + bot, store, tg := newTestBot([]openai.Component{{ID: "c-api", Name: "API", Status: "operational"}}) + store.subscribed = true + store.sub = redisstore.NewSubscriber(123, nil) + store.types = []string{redisstore.SubscriptionTypeIncident} + + bot.handleMessage(context.Background(), messageText("/subscribe component api")) + + if len(store.components) != 1 || store.components[0] != "c-api" { + t.Fatalf("components = %v, want [c-api]", store.components) + } + if !containsComponent(store.types, redisstore.SubscriptionTypeComponent) { + t.Fatalf("types = %v, want component enabled", store.types) + } + if len(tg.replies) != 1 || !strings.Contains(tg.replies[0], "Subscribed to component") { + t.Fatalf("reply = %v", tg.replies) + } +} + +func TestSubscribeComponentAllEnablesComponentType(t *testing.T) { + bot, store, _ := newTestBot(nil) + store.subscribed = true + store.sub = redisstore.NewSubscriber(123, nil) + store.types = []string{redisstore.SubscriptionTypeIncident} + store.components = []string{"c-api"} + + bot.handleMessage(context.Background(), messageText("/subscribe component all")) + + if len(store.components) != 0 { + t.Fatalf("components = %v, want cleared", store.components) + } + if !containsComponent(store.types, redisstore.SubscriptionTypeComponent) { + t.Fatalf("types = %v, want component enabled", store.types) + } +} + +func TestStatusAmbiguousComponentNameRequiresID(t *testing.T) { + bot, _, tg := newTestBot([]openai.Component{ + {ID: "login-a", Name: "Login", Status: "operational"}, + {ID: "login-b", Name: "Login", Status: "operational"}, + }) + + bot.handleMessage(context.Background(), messageText("/status Login")) + + if len(tg.replies) != 1 || !strings.Contains(tg.replies[0], "ambiguous") || !strings.Contains(tg.replies[0], "login-a") { + t.Fatalf("reply = %v", tg.replies) + } +} + +func TestCommandForOtherBotIsIgnored(t *testing.T) { + bot, _, tg := newTestBot(nil) + + bot.handleMessage(context.Background(), messageText("/start@OtherBot")) + + if len(tg.replies) != 0 { + t.Fatalf("replies = %v, want none", tg.replies) + } +} + +func newTestBot(components []openai.Component) (*Bot, *fakeBotStore, *fakeTelegramClient) { + tg := &fakeTelegramClient{} + store := &fakeBotStore{} + statusClient := fakeBotStatusClient{summary: openai.Summary{Components: components}} + return New(tg, statusClient, store, slog.Default(), "OpenAIStatusBot"), store, tg +} + +func messageText(text string) telegram.Message { + return telegram.Message{Text: text, Chat: telegram.Chat{ID: 123}} +} diff --git a/internal/bot/command_parse.go b/internal/bot/command_parse.go new file mode 100644 index 0000000..ee3dd97 --- /dev/null +++ b/internal/bot/command_parse.go @@ -0,0 +1,19 @@ +package bot + +import "strings" + +func normalizeCommand(text, botUsername string) (string, []string) { + fields := strings.Fields(strings.TrimSpace(text)) + if len(fields) == 0 { + return "", nil + } + command := strings.ToLower(fields[0]) + if at := strings.Index(command, "@"); at >= 0 { + target := command[at+1:] + if botUsername != "" && !strings.EqualFold(target, botUsername) { + return "", fields + } + command = command[:at] + } + return command, fields +} diff --git a/internal/bot/commands.go b/internal/bot/commands.go new file mode 100644 index 0000000..6951dfc --- /dev/null +++ b/internal/bot/commands.go @@ -0,0 +1,160 @@ +package bot + +import ( + "context" + "fmt" + "strings" + + "github.com/tiennm99/openai-status-bot/internal/redisstore" + "github.com/tiennm99/openai-status-bot/internal/telegram" +) + +func (b *Bot) replySubscribe(ctx context.Context, message telegram.Message, fields []string) { + sub := redisstore.NewSubscriber(message.Chat.ID, message.MessageThreadID) + current, subscribed, err := b.store.GetSubscriber(ctx, sub) + if err != nil { + b.logger.Error("get subscription", "error", err) + b.reply(ctx, message, "Could not load subscription right now.") + return + } + if len(fields) < 2 { + b.reply(ctx, message, formatSubscribeUsage(currentOrTarget(current, sub, subscribed), subscribed)) + return + } + + args := fields[1:] + arg := strings.ToLower(args[0]) + if arg == redisstore.SubscriptionTypeComponent && len(args) > 1 { + b.updateComponentFilter(ctx, message, sub, strings.Join(args[1:], " "), subscribed) + return + } + + var types []string + switch arg { + case redisstore.SubscriptionTypeIncident: + types = []string{redisstore.SubscriptionTypeIncident} + case redisstore.SubscriptionTypeComponent: + types = []string{redisstore.SubscriptionTypeComponent} + case "all": + types = redisstore.DefaultSubscriptionTypes() + default: + b.reply(ctx, message, formatSubscribeUsage(currentOrTarget(current, sub, subscribed), subscribed)) + return + } + + updated, err := b.store.UpdateSubscriberTypes(ctx, sub, types) + if err != nil { + b.logger.Error("update subscription", "error", err) + b.reply(ctx, message, "Could not update subscription right now.") + return + } + if !updated { + b.reply(ctx, message, "Not subscribed yet. Use /start first.") + return + } + b.reply(ctx, message, fmt.Sprintf("Subscription updated: %s", escape(strings.Join(types, ", ")))) +} + +func (b *Bot) updateComponentFilter(ctx context.Context, message telegram.Message, sub redisstore.Subscriber, componentArg string, subscribed bool) { + if !subscribed { + b.reply(ctx, message, "Not subscribed yet. Use /start first.") + return + } + componentArg = strings.TrimSpace(componentArg) + if strings.EqualFold(componentArg, "all") { + current, _, err := b.store.GetSubscriber(ctx, sub) + if err != nil { + b.logger.Error("load subscription", "error", err) + b.reply(ctx, message, "Could not update component filter right now.") + return + } + if _, err := b.store.UpdateSubscriberComponents(ctx, sub, nil); err != nil { + b.logger.Error("clear component filter", "error", err) + b.reply(ctx, message, "Could not update component filter right now.") + return + } + if !containsComponent(current.Types, redisstore.SubscriptionTypeComponent) { + current.Types = append(current.Types, redisstore.SubscriptionTypeComponent) + if _, err := b.store.UpdateSubscriberTypes(ctx, sub, current.Types); err != nil { + b.logger.Error("update component type", "error", err) + b.reply(ctx, message, "Could not update subscription right now.") + return + } + } + b.reply(ctx, message, "Component filter cleared. Receiving all component updates.") + return + } + + summary, err := b.statusClient.FetchSummary(ctx) + if err != nil { + b.logger.Error("fetch components", "error", err) + b.reply(ctx, message, "Could not fetch OpenAI components right now.") + return + } + resolution := resolveComponent(summary.Components, componentArg) + if !resolution.Found { + b.reply(ctx, message, fmt.Sprintf("Component %s not found.", escape(componentArg))) + return + } + if resolution.Ambiguous { + b.reply(ctx, message, formatAmbiguousComponents(componentArg, resolution.Matches)) + return + } + component := resolution.Component + + current, _, err := b.store.GetSubscriber(ctx, sub) + if err != nil { + b.logger.Error("load component filters", "error", err) + b.reply(ctx, message, "Could not update component filter right now.") + return + } + components := append([]string{}, current.Components...) + if !containsComponent(components, component.ID) { + components = append(components, component.ID) + } + if _, err := b.store.UpdateSubscriberComponents(ctx, sub, components); err != nil { + b.logger.Error("update component filter", "error", err) + b.reply(ctx, message, "Could not update component filter right now.") + return + } + if !containsComponent(current.Types, redisstore.SubscriptionTypeComponent) { + current.Types = append(current.Types, redisstore.SubscriptionTypeComponent) + if _, err := b.store.UpdateSubscriberTypes(ctx, sub, current.Types); err != nil { + b.logger.Error("update component type", "error", err) + b.reply(ctx, message, "Could not update subscription right now.") + 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)))) +} + +func (b *Bot) replyHistory(ctx context.Context, message telegram.Message, count int) { + incidents, err := b.statusClient.FetchIncidents(ctx) + if err != nil { + b.logger.Error("fetch incidents", "error", err) + b.reply(ctx, message, "Could not fetch OpenAI incident history right now.") + return + } + b.reply(ctx, message, formatHistory(incidents.Incidents, count)) +} + +func (b *Bot) replyUptime(ctx context.Context, message telegram.Message) { + summary, err := b.statusClient.FetchSummary(ctx) + if err != nil { + b.logger.Error("fetch uptime", "error", err) + b.reply(ctx, message, "Could not fetch OpenAI uptime right now.") + return + } + b.reply(ctx, message, formatUptime(summary)) +} + +func (b *Bot) replyInfo(ctx context.Context, message telegram.Message) { + sub := redisstore.NewSubscriber(message.Chat.ID, message.MessageThreadID) + current, subscribed, err := b.store.GetSubscriber(ctx, sub) + if err != nil { + b.logger.Error("get subscription info", "error", err) + b.reply(ctx, message, "Could not load subscription right now.") + return + } + b.reply(ctx, message, formatInfo(currentOrTarget(current, sub, subscribed), subscribed)) +} diff --git a/internal/bot/component_resolution.go b/internal/bot/component_resolution.go new file mode 100644 index 0000000..46993e2 --- /dev/null +++ b/internal/bot/component_resolution.go @@ -0,0 +1,121 @@ +package bot + +import ( + "fmt" + "strings" + + openai "github.com/tiennm99/openai-status-bot/internal/openai" +) + +type componentResolution struct { + Component openai.Component + Matches []openai.Component + Found bool + Ambiguous bool +} + +func resolveComponent(components []openai.Component, query string) componentResolution { + query = strings.TrimSpace(query) + if query == "" { + return componentResolution{} + } + for _, component := range components { + if component.Group { + continue + } + if strings.EqualFold(component.ID, query) { + return componentResolution{Component: component, Found: true} + } + } + + exact := matchingComponents(components, query, func(name, query string) bool { + return strings.EqualFold(name, query) + }) + if len(exact) == 1 { + return componentResolution{Component: exact[0], Found: true} + } + if len(exact) > 1 { + return componentResolution{Matches: exact, Found: true, Ambiguous: true} + } + + contains := matchingComponents(components, query, func(name, query string) bool { + return strings.Contains(strings.ToLower(name), strings.ToLower(query)) + }) + if len(contains) == 1 { + return componentResolution{Component: contains[0], Found: true} + } + if len(contains) > 1 { + return componentResolution{Matches: contains, Found: true, Ambiguous: true} + } + return componentResolution{} +} + +func matchingComponents(components []openai.Component, query string, matches func(name, query string) bool) []openai.Component { + result := make([]openai.Component, 0) + for _, component := range components { + if component.Group { + continue + } + if matches(component.Name, query) { + result = append(result, component) + } + } + return result +} + +func formatAmbiguousComponents(query string, matches []openai.Component) string { + duplicates := 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))) + } + return truncateMessage(strings.Join(lines, "\n")) +} + +func componentFilterLabels(components []openai.Component, ids []string) string { + if len(ids) == 0 { + return "all" + } + duplicates := 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]) + break + } + } + labels = append(labels, label) + } + 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 5aab1b1..22150bf 100644 --- a/internal/bot/format.go +++ b/internal/bot/format.go @@ -2,35 +2,44 @@ package bot import ( "fmt" + "html" "strconv" "strings" + "time" openai "github.com/tiennm99/openai-status-bot/internal/openai" "github.com/tiennm99/openai-status-bot/internal/poller" ) -const helpText = `OpenAI Status Bot +const statusURL = "https://status.openai.com/" + +const helpText = `OpenAI Status Bot /start - subscribe this chat or topic /stop - unsubscribe this chat or topic -/status - show current OpenAI status +/status [component] - show current OpenAI status /components - show all component statuses +/subscribe <incident|component|all> - set notification types +/subscribe component <name|all> - filter component updates /history [count] - show recent incidents, default 5, max 10 +/uptime - show component health overview +/info - show chat and subscription details /help - show this help` func formatStatus(summary openai.Summary) string { lines := []string{ - "OpenAI status", + "OpenAI Status", "", - fmt.Sprintf("Overall: %s", summary.Status.Description), + fmt.Sprintf("Overall: %s", escape(summary.Status.Description)), } + duplicates := 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", component.Name, poller.StatusLabel(component.Status))) + degraded = append(degraded, fmt.Sprintf("- %s: %s", escape(componentLabel(component, duplicates[component.Name])), escape(poller.StatusLabel(component.Status)))) } if len(degraded) == 0 { @@ -39,18 +48,29 @@ func formatStatus(summary openai.Summary) string { lines = append(lines, "", "Affected components:") lines = append(lines, degraded...) } - lines = append(lines, "", "https://status.openai.com/") + lines = append(lines, "", fmt.Sprintf("View full status page", statusURL)) return strings.Join(lines, "\n") } +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.StatusLabel(component.Status)), + escape(formatTime(component.UpdatedAt)), + statusURL, + ) +} + func formatComponents(summary openai.Summary) string { - lines := []string{"OpenAI components", ""} + lines := []string{"OpenAI Components", ""} + duplicates := duplicateComponentNames(summary.Components) for _, component := range summary.Components { if component.Group { continue } - lines = append(lines, fmt.Sprintf("- %s: %s", component.Name, poller.StatusLabel(component.Status))) + lines = append(lines, fmt.Sprintf("- %s: %s", escape(componentLabel(component, duplicates[component.Name])), escape(poller.StatusLabel(component.Status)))) } return truncateMessage(strings.Join(lines, "\n")) } @@ -63,17 +83,46 @@ func formatHistory(incidents []openai.Incident, count int) string { count = len(incidents) } - lines := []string{"Recent OpenAI incidents", ""} + lines := []string{"Recent OpenAI Incidents", ""} for i := 0; i < count; i++ { incident := incidents[i] - lines = append(lines, fmt.Sprintf( - "%d. %s\n Status: %s | Impact: %s", + link := incident.Shortlink + if link == "" && incident.ID != "" { + link = statusURL + "incidents/" + incident.ID + } + entry := fmt.Sprintf( + "%d. [%s] %s\n Created: %s UTC\n Status: %s", i+1, - incident.Name, - poller.StatusLabel(incident.Status), - poller.StatusLabel(incident.Impact), + escape(strings.ToUpper(emptyDefault(incident.Impact, "unknown"))), + escape(incident.Name), + escape(formatDate(incident.CreatedAt)), + escape(poller.StatusLabel(incident.Status)), + ) + if link != "" { + entry += fmt.Sprintf("\n Details", escapeAttr(link)) + } + lines = append(lines, entry) + } + lines = append(lines, "", fmt.Sprintf("View full history", statusURL)) + return truncateMessage(strings.Join(lines, "\n\n")) +} + +func formatUptime(summary openai.Summary) string { + lines := []string{"OpenAI Component Health", ""} + duplicates := 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.StatusLabel(component.Status)), + escape(formatTime(component.UpdatedAt)), )) } + lines = append(lines, "", "Uptime percentage is not available from the public Statuspage API.") + lines = append(lines, fmt.Sprintf("View full status page", statusURL)) return truncateMessage(strings.Join(lines, "\n")) } @@ -95,22 +144,42 @@ func parseHistoryCount(fields []string) int { return count } -func normalizeCommand(text string) (string, []string) { - fields := strings.Fields(strings.TrimSpace(text)) - if len(fields) == 0 { - return "", nil - } - command := strings.ToLower(fields[0]) - if at := strings.Index(command, "@"); at >= 0 { - command = command[:at] - } - return command, fields -} - func truncateMessage(value string) string { const telegramLimit = 3900 - if len(value) <= telegramLimit { + runes := []rune(value) + if len(runes) <= telegramLimit { return value } - return strings.TrimSpace(value[:telegramLimit-3]) + "..." + return strings.TrimSpace(string(runes[:telegramLimit-3])) + "..." +} + +func formatDate(value string) string { + parsed, err := time.Parse(time.RFC3339, value) + if err != nil { + return emptyDefault(value, "unknown") + } + return parsed.UTC().Format("Jan 2, 2006") +} + +func formatTime(value string) string { + parsed, err := time.Parse(time.RFC3339, value) + if err != nil { + return emptyDefault(value, "unknown") + } + return parsed.UTC().Format("Jan 2, 2006 15:04") +} + +func emptyDefault(value, fallback string) string { + if strings.TrimSpace(value) == "" { + return fallback + } + return value +} + +func escape(value string) string { + return html.EscapeString(value) +} + +func escapeAttr(value string) string { + return html.EscapeString(value) } diff --git a/internal/bot/format_test.go b/internal/bot/format_test.go index 79b6e3f..ab9ca93 100644 --- a/internal/bot/format_test.go +++ b/internal/bot/format_test.go @@ -2,8 +2,8 @@ package bot import "testing" -func TestNormalizeCommandStripsBotUsername(t *testing.T) { - command, fields := normalizeCommand("/history@OpenAIStatusBot 10") +func TestNormalizeCommandStripsOwnBotUsername(t *testing.T) { + command, fields := normalizeCommand("/history@OpenAIStatusBot 10", "OpenAIStatusBot") if command != "/history" { t.Fatalf("command = %q, want /history", command) } @@ -12,6 +12,13 @@ func TestNormalizeCommandStripsBotUsername(t *testing.T) { } } +func TestNormalizeCommandIgnoresOtherBotUsername(t *testing.T) { + command, _ := normalizeCommand("/start@OtherBot", "OpenAIStatusBot") + if command != "" { + t.Fatalf("command = %q, want empty", command) + } +} + func TestParseHistoryCount(t *testing.T) { tests := []struct { name string diff --git a/internal/bot/helpers.go b/internal/bot/helpers.go new file mode 100644 index 0000000..e175555 --- /dev/null +++ b/internal/bot/helpers.go @@ -0,0 +1,67 @@ +package bot + +import ( + "context" + "fmt" + "strings" + + "github.com/tiennm99/openai-status-bot/internal/redisstore" + "github.com/tiennm99/openai-status-bot/internal/telegram" +) + +func (b *Bot) replyStatus(ctx context.Context, message telegram.Message, fields []string) { + summary, err := b.statusClient.FetchSummary(ctx) + if err != nil { + b.logger.Error("fetch status", "error", err) + b.reply(ctx, message, "Could not fetch OpenAI status right now.") + return + } + if len(fields) > 1 { + query := strings.Join(fields[1:], " ") + resolution := resolveComponent(summary.Components, query) + if !resolution.Found { + b.reply(ctx, message, fmt.Sprintf("Component %s not found.", escape(query))) + return + } + if resolution.Ambiguous { + b.reply(ctx, message, formatAmbiguousComponents(query, resolution.Matches)) + return + } + duplicates := duplicateComponentNames(summary.Components) + b.reply(ctx, message, formatComponentStatus(resolution.Component, duplicates[resolution.Component.Name])) + return + } + b.reply(ctx, message, formatStatus(summary)) +} + +func (b *Bot) replyComponents(ctx context.Context, message telegram.Message) { + summary, err := b.statusClient.FetchSummary(ctx) + if err != nil { + b.logger.Error("fetch components", "error", err) + b.reply(ctx, message, "Could not fetch OpenAI components right now.") + return + } + b.reply(ctx, message, formatComponents(summary)) +} + +func (b *Bot) reply(ctx context.Context, message telegram.Message, text string) { + if err := b.telegramClient.SendText(ctx, message.Chat.ID, message.MessageThreadID, text); err != nil { + b.logger.Warn("send telegram reply", "chat_id", message.Chat.ID, "error", err) + } +} + +func currentOrTarget(current, target redisstore.Subscriber, subscribed bool) redisstore.Subscriber { + if subscribed { + return current + } + return target +} + +func containsComponent(values []string, target string) bool { + for _, value := range values { + if strings.EqualFold(value, target) { + return true + } + } + return false +} diff --git a/internal/bot/subscription_format.go b/internal/bot/subscription_format.go new file mode 100644 index 0000000..dc34187 --- /dev/null +++ b/internal/bot/subscription_format.go @@ -0,0 +1,45 @@ +package bot + +import ( + "fmt" + "strconv" + "strings" + + "github.com/tiennm99/openai-status-bot/internal/redisstore" +) + +func formatInfo(sub redisstore.Subscriber, subscribed bool) string { + chatID := strconv.FormatInt(sub.ChatID, 10) + if sub.ThreadID != nil { + chatID = fmt.Sprintf("%s:%d", chatID, *sub.ThreadID) + } + if !subscribed { + return fmt.Sprintf("Chat Info\n\nChat ID: %s\n\nNot subscribed. Use /start to subscribe.", escape(chatID)) + } + components := "all" + if len(sub.Components) > 0 { + components = strings.Join(sub.Components, ", ") + } + return fmt.Sprintf( + "Chat Info\n\nChat ID: %s\n\nTypes: %s\nComponents: %s", + escape(chatID), + escape(strings.Join(sub.Types, ", ")), + escape(components), + ) +} + +func formatSubscribeUsage(sub redisstore.Subscriber, subscribed bool) string { + current := "none (use /start first)" + components := "all" + if subscribed { + current = strings.Join(sub.Types, ", ") + if len(sub.Components) > 0 { + components = strings.Join(sub.Components, ", ") + } + } + return fmt.Sprintf( + "Usage: /subscribe <type> [component]\n\nTypes: incident, component, all\nComponent filter: /subscribe component api\nClear filter: /subscribe component all\n\nCurrent types: %s\nComponents: %s", + escape(current), + escape(components), + ) +} diff --git a/internal/openai/models.go b/internal/openai/models.go index 0125a89..a93f53d 100644 --- a/internal/openai/models.go +++ b/internal/openai/models.go @@ -20,6 +20,7 @@ type Component struct { UpdatedAt string `json:"updated_at"` Position int `json:"position"` PageID string `json:"page_id"` + GroupID string `json:"group_id"` Group bool `json:"group"` } @@ -38,6 +39,7 @@ type Incident struct { UpdatedAt string `json:"updated_at"` ResolvedAt string `json:"resolved_at"` Impact string `json:"impact"` + Shortlink string `json:"shortlink"` PageID string `json:"page_id"` IncidentUpdates []IncidentUpdate `json:"incident_updates"` } diff --git a/internal/poller/delivery.go b/internal/poller/delivery.go new file mode 100644 index 0000000..feaadf0 --- /dev/null +++ b/internal/poller/delivery.go @@ -0,0 +1,47 @@ +package poller + +import ( + "context" + + "github.com/tiennm99/openai-status-bot/internal/redisstore" + "github.com/tiennm99/openai-status-bot/internal/telegram" +) + +func (r *Runner) notifySubscribers(ctx context.Context, event notificationEvent, subscribers []redisstore.Subscriber, removed map[string]bool) error { + for _, subscriber := range subscribers { + subscriberKey := subscriber.Key() + if removed[subscriberKey] { + continue + } + 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 err + } + if delivered { + continue + } + } + if err := r.notifier.SendMessage(ctx, subscriber, event.text); err != nil { + if telegram.IsTerminalSendError(err) { + if removeErr := r.store.RemoveSubscriber(ctx, subscriber); removeErr != nil { + return removeErr + } + removed[subscriberKey] = true + r.logger.Info("removed unreachable telegram subscriber", "subscriber", subscriberKey, "error", err) + continue + } + r.logger.Warn("send telegram message", "subscriber", subscriberKey, "error", err) + return err + } + if event.deliveryKey != "" { + if err := r.store.MarkDelivered(ctx, event.deliveryKey, subscriberKey); err != nil { + return err + } + } + } + return nil +} diff --git a/internal/poller/event_collection.go b/internal/poller/event_collection.go new file mode 100644 index 0000000..c94ad63 --- /dev/null +++ b/internal/poller/event_collection.go @@ -0,0 +1,198 @@ +package poller + +import ( + "context" + "crypto/sha256" + "fmt" + "sort" + "strings" + + openai "github.com/tiennm99/openai-status-bot/internal/openai" + "github.com/tiennm99/openai-status-bot/internal/redisstore" +) + +func (r *Runner) collectEvents(ctx context.Context, summary openai.Summary, incidents openai.IncidentsResponse, initialized bool) ([]notificationEvent, []checkpoint, []checkpoint, error) { + componentEvents, componentBefore, componentAfter, err := r.collectComponentEvents(ctx, summary, initialized) + if err != nil { + return nil, nil, nil, err + } + incidentEvents, incidentAfter, err := r.collectIncidentEvents(ctx, incidents, initialized) + if err != nil { + return nil, nil, nil, err + } + + events := append(componentEvents, incidentEvents...) + sort.SliceStable(events, func(i, j int) bool { + if events[i].sortTime == "" { + return false + } + if events[j].sortTime == "" { + return true + } + return events[i].sortTime < events[j].sortTime + }) + after := append(componentAfter, incidentAfter...) + return events, componentBefore, after, nil +} + +func (r *Runner) collectComponentEvents(ctx context.Context, summary openai.Summary, initialized bool) ([]notificationEvent, []checkpoint, []checkpoint, error) { + knownStatuses, err := r.store.ComponentStatuses(ctx) + if err != nil { + return nil, nil, nil, err + } + pending, err := r.store.PendingComponentEvents(ctx) + if err != nil { + return nil, nil, nil, err + } + + componentsForLabels := append([]openai.Component{}, summary.Components...) + for _, pendingEvent := range pending { + componentsForLabels = append(componentsForLabels, pendingComponent(pendingEvent)) + } + duplicates := duplicateComponentNames(componentsForLabels) + + events := make([]notificationEvent, 0) + before := make([]checkpoint, 0) + after := make([]checkpoint, 0, len(summary.Components)+len(pending)) + + for _, pendingEvent := range pending { + pendingEvent := pendingEvent + component := pendingComponent(pendingEvent) + events = append(events, notificationEvent{ + eventType: redisstore.SubscriptionTypeComponent, + componentID: pendingEvent.ComponentID, + componentName: pendingEvent.ComponentName, + deliveryKey: pendingEvent.DeliveryKey, + 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) + }) + } + + for _, component := range summary.Components { + if component.ID == "" || component.Group { + continue + } + component := component + if _, hasPending := pending[component.ID]; hasPending { + continue + } + + previousStatus, found := knownStatuses[component.ID] + if !initialized { + after = append(after, func(ctx context.Context) error { + return r.store.SaveComponentStatus(ctx, component.ID, component.Status) + }) + continue + } + if found && previousStatus == component.Status { + after = append(after, func(ctx context.Context) error { + return r.store.SaveComponentStatus(ctx, component.ID, component.Status) + }) + continue + } + if !found { + previousStatus = "unknown" + } + + deliveryKey := fmt.Sprintf("component:%s:%s:%s", component.ID, component.Status, component.UpdatedAt) + pendingEvent := redisstore.PendingComponentEvent{ + ComponentID: component.ID, + ComponentName: component.Name, + Status: component.Status, + UpdatedAt: component.UpdatedAt, + Position: component.Position, + PreviousStatus: previousStatus, + DeliveryKey: deliveryKey, + } + before = append(before, func(ctx context.Context) error { + return r.store.SavePendingComponentEvent(ctx, pendingEvent) + }) + events = append(events, notificationEvent{ + eventType: redisstore.SubscriptionTypeComponent, + componentID: component.ID, + componentName: component.Name, + deliveryKey: deliveryKey, + 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) + }) + } + return events, before, after, nil +} + +func (r *Runner) collectIncidentEvents(ctx context.Context, response openai.IncidentsResponse, initialized bool) ([]notificationEvent, []checkpoint, error) { + events := make([]notificationEvent, 0) + checkpoints := make([]checkpoint, 0) + for _, incident := range response.Incidents { + incident := incident + for _, update := range incident.IncidentUpdates { + if update.ID == "" { + continue + } + update := update + version := IncidentUpdateVersion(update) + seen, err := r.store.HasIncidentUpdateVersion(ctx, update.ID, version) + if err != nil { + return nil, nil, err + } + checkpoints = append(checkpoints, func(ctx context.Context) error { + return r.store.MarkIncidentUpdateVersion(ctx, update.ID, version) + }) + deliveryKey := fmt.Sprintf("incident:%s:%s", update.ID, version) + if initialized && !seen { + events = append(events, notificationEvent{ + eventType: redisstore.SubscriptionTypeIncident, + deliveryKey: deliveryKey, + sortTime: incidentUpdateSortTime(update), + text: FormatIncidentUpdate(incident, update), + }) + checkpoints = append(checkpoints, func(ctx context.Context) error { + return r.store.ClearDelivery(ctx, deliveryKey) + }) + } + } + } + return events, checkpoints, nil +} + +func pendingComponent(event redisstore.PendingComponentEvent) openai.Component { + return openai.Component{ + ID: event.ComponentID, + Name: event.ComponentName, + Status: event.Status, + UpdatedAt: event.UpdatedAt, + Position: event.Position, + } +} + +func IncidentUpdateVersion(update openai.IncidentUpdate) string { + parts := []string{update.UpdatedAt, update.Status, update.Body, update.DisplayAt, update.CreatedAt} + sum := sha256.Sum256([]byte(strings.Join(parts, "\x00"))) + return fmt.Sprintf("%x", sum[:]) +} + +func incidentUpdateSortTime(update openai.IncidentUpdate) string { + for _, value := range []string{update.DisplayAt, update.CreatedAt, update.UpdatedAt} { + if value != "" { + return value + } + } + return "" +} diff --git a/internal/poller/format.go b/internal/poller/format.go index 62819bc..915a32c 100644 --- a/internal/poller/format.go +++ b/internal/poller/format.go @@ -2,18 +2,18 @@ package poller import ( "fmt" + "html" "strings" openai "github.com/tiennm99/openai-status-bot/internal/openai" ) -func FormatComponentChange(component openai.Component, previousStatus string) string { +func FormatComponentChange(component openai.Component, previousStatus string, duplicate bool) string { return fmt.Sprintf( - "OpenAI component status changed\n\nComponent: %s\nStatus: %s -> %s\n\n%s", - component.Name, - StatusLabel(previousStatus), - StatusLabel(component.Status), - "https://status.openai.com/", + "OpenAI component update\n\nComponent: %s\nStatus: %s -> %s\n\nView full status page", + escape(componentLabel(component, duplicate)), + escape(StatusLabel(previousStatus)), + escape(StatusLabel(component.Status)), ) } @@ -23,13 +23,18 @@ func FormatIncidentUpdate(incident openai.Incident, update openai.IncidentUpdate body = "No update message provided." } + link := incident.Shortlink + if link == "" && incident.ID != "" { + link = "https://status.openai.com/incidents/" + incident.ID + } + return fmt.Sprintf( - "OpenAI incident update\n\n%s\nStatus: %s\nImpact: %s\n\n%s\n\n%s", - incident.Name, - StatusLabel(update.Status), - StatusLabel(incident.Impact), - truncate(body, 2600), - "https://status.openai.com/incidents/"+incident.ID, + "OpenAI incident update\n\n%s\nStatus: %s\nImpact: %s\n\n%s\n\nView incident", + escape(incident.Name), + escape(StatusLabel(update.Status)), + escape(StatusLabel(incident.Impact)), + escape(truncate(body, 2600)), + escape(link), ) } @@ -40,13 +45,13 @@ func StatusLabel(value string) string { case "operational": return "Operational" case "degraded_performance": - return "Degraded performance" + return "Degraded Performance" case "partial_outage": - return "Partial outage" + return "Partial Outage" case "major_outage": - return "Major outage" + return "Major Outage" case "under_maintenance": - return "Under maintenance" + return "Under Maintenance" case "none": return "None" case "minor": @@ -55,6 +60,8 @@ func StatusLabel(value string) string { return "Major" case "critical": return "Critical" + case "maintenance": + return "Maintenance" case "investigating": return "Investigating" case "identified": @@ -68,12 +75,46 @@ func StatusLabel(value string) string { } } +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] +} + func truncate(value string, limit int) string { - if len(value) <= limit { + runes := []rune(value) + if len(runes) <= limit { return value } if limit <= 3 { - return value[:limit] + return string(runes[:limit]) } - return strings.TrimSpace(value[:limit-3]) + "..." + return strings.TrimSpace(string(runes[:limit-3])) + "..." +} + +func escape(value string) string { + return html.EscapeString(value) } diff --git a/internal/poller/format_test.go b/internal/poller/format_test.go index f0c5d04..0eefb59 100644 --- a/internal/poller/format_test.go +++ b/internal/poller/format_test.go @@ -10,8 +10,8 @@ import ( func TestStatusLabel(t *testing.T) { tests := map[string]string{ "operational": "Operational", - "degraded_performance": "Degraded performance", - "partial_outage": "Partial outage", + "degraded_performance": "Degraded Performance", + "partial_outage": "Partial Outage", "minor": "Minor", "": "Unknown", } diff --git a/internal/poller/poller.go b/internal/poller/poller.go index 079f015..e015f06 100644 --- a/internal/poller/poller.go +++ b/internal/poller/poller.go @@ -16,10 +16,17 @@ type StatusClient interface { type Store interface { ComponentStatuses(ctx context.Context) (map[string]string, error) - HasIncidentUpdate(ctx context.Context, updateID string) (bool, error) + ClearDelivery(ctx context.Context, eventKey string) error + HasDelivered(ctx context.Context, eventKey, subscriberKey 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) - MarkIncidentUpdate(ctx context.Context, updateID string) error + MarkDelivered(ctx context.Context, eventKey, subscriberKey string) error + MarkIncidentUpdateVersion(ctx context.Context, updateID, version string) error + PendingComponentEvents(ctx context.Context) (map[string]redisstore.PendingComponentEvent, error) + RemovePendingComponentEvent(ctx context.Context, componentID string) error + RemoveSubscriber(ctx context.Context, sub redisstore.Subscriber) error + SavePendingComponentEvent(ctx context.Context, event redisstore.PendingComponentEvent) error SaveComponentStatus(ctx context.Context, componentID, status string) error SetInitialized(ctx context.Context) error } @@ -36,6 +43,17 @@ type Runner struct { logger *slog.Logger } +type notificationEvent struct { + eventType string + componentID string + componentName string + deliveryKey string + sortTime string + text string +} + +type checkpoint func(ctx context.Context) error + func NewRunner(statusClient StatusClient, store Store, notifier Notifier, interval time.Duration, logger *slog.Logger) *Runner { return &Runner{ statusClient: statusClient, @@ -77,18 +95,34 @@ func (r *Runner) CheckOnce(ctx context.Context) error { return err } - messages, err := r.collectComponentMessages(ctx, summary, initialized) + events, beforeDelivery, checkpoints, err := r.collectEvents(ctx, summary, incidents, initialized) if err != nil { return err } - incidentMessages, err := r.collectIncidentMessages(ctx, incidents, initialized) - if err != nil { - return err - } - messages = append(messages, incidentMessages...) - if initialized && len(messages) > 0 { - r.notifyAll(ctx, messages) + for _, save := range beforeDelivery { + if err := save(ctx); err != nil { + return err + } + } + + if initialized && len(events) > 0 { + subscribers, err := r.store.ListSubscribers(ctx) + if err != nil { + return err + } + removedSubscribers := map[string]bool{} + for _, event := range events { + if err := r.notifySubscribers(ctx, event, subscribers, removedSubscribers); err != nil { + return err + } + } + } + + for _, save := range checkpoints { + if err := save(ctx); err != nil { + return err + } } if !initialized { if err := r.store.SetInitialized(ctx); err != nil { @@ -99,72 +133,6 @@ func (r *Runner) CheckOnce(ctx context.Context) error { return nil } -func (r *Runner) collectComponentMessages(ctx context.Context, summary openai.Summary, initialized bool) ([]string, error) { - knownStatuses, err := r.store.ComponentStatuses(ctx) - if err != nil { - return nil, err - } - - messages := make([]string, 0) - for _, component := range summary.Components { - if component.ID == "" || component.Group { - continue - } - previousStatus, found := knownStatuses[component.ID] - if initialized && found && previousStatus != component.Status { - messages = append(messages, FormatComponentChange(component, previousStatus)) - } - if err := r.store.SaveComponentStatus(ctx, component.ID, component.Status); err != nil { - return nil, err - } - } - return messages, nil -} - -func (r *Runner) collectIncidentMessages(ctx context.Context, response openai.IncidentsResponse, initialized bool) ([]string, error) { - messages := make([]string, 0) - for _, incident := range response.Incidents { - for i := len(incident.IncidentUpdates) - 1; i >= 0; i-- { - update := incident.IncidentUpdates[i] - if update.ID == "" { - continue - } - seen, err := r.store.HasIncidentUpdate(ctx, update.ID) - if err != nil { - return nil, err - } - if initialized && !seen { - messages = append(messages, FormatIncidentUpdate(incident, update)) - } - if !seen { - if err := r.store.MarkIncidentUpdate(ctx, update.ID); err != nil { - return nil, err - } - } - } - } - return messages, nil -} - -func (r *Runner) notifyAll(ctx context.Context, messages []string) { - subscribers, err := r.store.ListSubscribers(ctx) - if err != nil { - r.logger.Error("list subscribers", "error", err) - return - } - if len(subscribers) == 0 { - return - } - - for _, message := range messages { - for _, subscriber := range subscribers { - if err := r.notifier.SendMessage(ctx, subscriber, message); err != nil { - r.logger.Warn("send telegram message", "subscriber", subscriber.Key(), "error", err) - } - } - } -} - func (r *Runner) checkAndLog(ctx context.Context) { if err := r.CheckOnce(ctx); err != nil && ctx.Err() == nil { r.logger.Error("poll openai status", "error", err) diff --git a/internal/poller/poller_test.go b/internal/poller/poller_test.go new file mode 100644 index 0000000..98c4680 --- /dev/null +++ b/internal/poller/poller_test.go @@ -0,0 +1,266 @@ +package poller + +import ( + "context" + "errors" + "log/slog" + "testing" + "time" + + openai "github.com/tiennm99/openai-status-bot/internal/openai" + "github.com/tiennm99/openai-status-bot/internal/redisstore" + "github.com/tiennm99/openai-status-bot/internal/telegram" +) + +type fakeStatusClient struct { + summary openai.Summary + incidents openai.IncidentsResponse +} + +func (f fakeStatusClient) FetchSummary(context.Context) (openai.Summary, error) { + return f.summary, nil +} + +func (f fakeStatusClient) FetchIncidents(context.Context) (openai.IncidentsResponse, error) { + return f.incidents, nil +} + +type fakePollerStore struct { + initialized bool + componentStatuses map[string]string + incidentVersions map[string]string + subscribers []redisstore.Subscriber + delivered map[string]map[string]bool + savedComponents map[string]string + markedVersions map[string]string + pendingComponents map[string]redisstore.PendingComponentEvent + removed []string +} + +func newFakePollerStore() *fakePollerStore { + return &fakePollerStore{ + componentStatuses: map[string]string{}, + incidentVersions: map[string]string{}, + delivered: map[string]map[string]bool{}, + savedComponents: map[string]string{}, + markedVersions: map[string]string{}, + pendingComponents: map[string]redisstore.PendingComponentEvent{}, + } +} + +func (f *fakePollerStore) ComponentStatuses(context.Context) (map[string]string, error) { + return f.componentStatuses, nil +} + +func (f *fakePollerStore) ClearDelivery(_ context.Context, eventKey string) error { + delete(f.delivered, eventKey) + return nil +} + +func (f *fakePollerStore) HasDelivered(_ context.Context, eventKey, subscriberKey string) (bool, error) { + return f.delivered[eventKey][subscriberKey], nil +} + +func (f *fakePollerStore) HasIncidentUpdateVersion(_ context.Context, updateID, version string) (bool, error) { + stored, ok := f.incidentVersions[updateID] + return ok && stored == version, nil +} + +func (f *fakePollerStore) IsInitialized(context.Context) (bool, error) { + return f.initialized, nil +} + +func (f *fakePollerStore) ListSubscribers(context.Context) ([]redisstore.Subscriber, error) { + return f.subscribers, nil +} + +func (f *fakePollerStore) MarkDelivered(_ context.Context, eventKey, subscriberKey string) error { + if f.delivered[eventKey] == nil { + f.delivered[eventKey] = map[string]bool{} + } + f.delivered[eventKey][subscriberKey] = true + return nil +} + +func (f *fakePollerStore) MarkIncidentUpdateVersion(_ context.Context, updateID, version string) error { + f.incidentVersions[updateID] = version + f.markedVersions[updateID] = version + return nil +} + +func (f *fakePollerStore) PendingComponentEvents(context.Context) (map[string]redisstore.PendingComponentEvent, error) { + result := make(map[string]redisstore.PendingComponentEvent, len(f.pendingComponents)) + for key, value := range f.pendingComponents { + result[key] = value + } + return result, nil +} + +func (f *fakePollerStore) SavePendingComponentEvent(_ context.Context, event redisstore.PendingComponentEvent) error { + f.pendingComponents[event.ComponentID] = event + return nil +} + +func (f *fakePollerStore) RemovePendingComponentEvent(_ context.Context, componentID string) error { + delete(f.pendingComponents, componentID) + return nil +} + +func (f *fakePollerStore) RemoveSubscriber(_ context.Context, sub redisstore.Subscriber) error { + f.removed = append(f.removed, sub.Key()) + return nil +} + +func (f *fakePollerStore) SaveComponentStatus(_ context.Context, componentID, status string) error { + f.componentStatuses[componentID] = status + f.savedComponents[componentID] = status + return nil +} + +func (f *fakePollerStore) SetInitialized(context.Context) error { + f.initialized = true + return nil +} + +type fakeNotifier struct { + errors map[string]error + sends []string +} + +func (f *fakeNotifier) SendMessage(_ context.Context, sub redisstore.Subscriber, _ string) error { + f.sends = append(f.sends, sub.Key()) + if f.errors != nil { + return f.errors[sub.Key()] + } + return nil +} + +func TestCheckOnceDoesNotCheckpointWhenSendFails(t *testing.T) { + store := newFakePollerStore() + store.initialized = true + store.componentStatuses["c1"] = "operational" + store.subscribers = []redisstore.Subscriber{redisstore.NewSubscriber(1, nil)} + notifier := &fakeNotifier{errors: map[string]error{"1": errors.New("network")}} + runner := NewRunner(fakeStatusClient{summary: summaryWithComponent("c1", "API", "degraded_performance")}, store, notifier, time.Minute, slog.Default()) + + if err := runner.CheckOnce(context.Background()); err == nil { + t.Fatal("expected send error") + } + if got := store.savedComponents["c1"]; got != "" { + t.Fatalf("component checkpoint = %q, want empty", got) + } +} + +func TestCheckOnceSkipsAlreadyDeliveredSubscriberOnRetry(t *testing.T) { + store := newFakePollerStore() + store.initialized = true + store.componentStatuses["c1"] = "operational" + store.subscribers = []redisstore.Subscriber{redisstore.NewSubscriber(1, nil), redisstore.NewSubscriber(2, nil)} + notifier := &fakeNotifier{errors: map[string]error{"2": errors.New("rate limit")}} + runner := NewRunner(fakeStatusClient{summary: summaryWithComponent("c1", "API", "degraded_performance")}, store, notifier, time.Minute, slog.Default()) + + if err := runner.CheckOnce(context.Background()); err == nil { + t.Fatal("expected first send error") + } + if len(notifier.sends) != 2 || notifier.sends[0] != "1" || notifier.sends[1] != "2" { + t.Fatalf("first sends = %v, want [1 2]", notifier.sends) + } + notifier.errors = nil + notifier.sends = nil + if err := runner.CheckOnce(context.Background()); err != nil { + t.Fatalf("second CheckOnce returned error: %v", err) + } + if len(notifier.sends) != 1 || notifier.sends[0] != "2" { + t.Fatalf("second sends = %v, want [2]", notifier.sends) + } + if got := store.savedComponents["c1"]; got != "degraded_performance" { + t.Fatalf("component checkpoint = %q", got) + } +} + +func TestCheckOnceRemovesTerminalSubscriberAndCheckpoints(t *testing.T) { + store := newFakePollerStore() + store.initialized = true + store.componentStatuses["c1"] = "operational" + store.subscribers = []redisstore.Subscriber{redisstore.NewSubscriber(1, nil), redisstore.NewSubscriber(2, nil)} + notifier := &fakeNotifier{errors: map[string]error{"2": &telegram.APIError{StatusCode: 403, Description: "Forbidden: bot was blocked by the user"}}} + runner := NewRunner(fakeStatusClient{summary: summaryWithComponent("c1", "API", "degraded_performance")}, store, notifier, time.Minute, slog.Default()) + + if err := runner.CheckOnce(context.Background()); err != nil { + t.Fatalf("CheckOnce returned error: %v", err) + } + if len(store.removed) != 1 || store.removed[0] != "2" { + t.Fatalf("removed = %v, want [2]", store.removed) + } + if got := store.savedComponents["c1"]; got != "degraded_performance" { + t.Fatalf("component checkpoint = %q", got) + } +} + +func TestIncidentUpdateVersionEditSendsAgain(t *testing.T) { + update := openai.IncidentUpdate{ID: "u1", Body: "old", UpdatedAt: "2026-01-01T00:00:00Z"} + store := newFakePollerStore() + store.initialized = true + store.incidentVersions["u1"] = IncidentUpdateVersion(update) + store.subscribers = []redisstore.Subscriber{redisstore.NewSubscriber(1, nil)} + updated := openai.IncidentUpdate{ID: "u1", Body: "new", UpdatedAt: "2026-01-01T00:01:00Z"} + notifier := &fakeNotifier{} + runner := NewRunner(fakeStatusClient{incidents: openai.IncidentsResponse{Incidents: []openai.Incident{{ID: "i1", Name: "incident", Impact: "minor", IncidentUpdates: []openai.IncidentUpdate{updated}}}}}, store, notifier, time.Minute, slog.Default()) + + if err := runner.CheckOnce(context.Background()); err != nil { + t.Fatalf("CheckOnce returned error: %v", err) + } + if len(notifier.sends) != 1 { + t.Fatalf("sends = %v, want one send", notifier.sends) + } + if got := store.markedVersions["u1"]; got != IncidentUpdateVersion(updated) { + t.Fatalf("marked version = %q", got) + } +} + +func TestPendingComponentEventSurvivesReturnToPreviousStatus(t *testing.T) { + store := newFakePollerStore() + store.initialized = true + store.componentStatuses["c1"] = "operational" + store.subscribers = []redisstore.Subscriber{redisstore.NewSubscriber(1, nil), redisstore.NewSubscriber(2, nil)} + notifier := &fakeNotifier{errors: map[string]error{"2": errors.New("rate limit")}} + runner := NewRunner(fakeStatusClient{summary: summaryWithComponent("c1", "API", "degraded_performance")}, store, notifier, time.Minute, slog.Default()) + + if err := runner.CheckOnce(context.Background()); err == nil { + t.Fatal("expected first send error") + } + if len(store.pendingComponents) != 1 { + t.Fatalf("pending components = %v, want one", store.pendingComponents) + } + + notifier.errors = nil + notifier.sends = nil + runner.statusClient = fakeStatusClient{summary: summaryWithComponent("c1", "API", "operational")} + if err := runner.CheckOnce(context.Background()); err != nil { + t.Fatalf("second CheckOnce returned error: %v", err) + } + if len(notifier.sends) != 1 || notifier.sends[0] != "2" { + t.Fatalf("second sends = %v, want retry only for subscriber 2", notifier.sends) + } + if got := store.savedComponents["c1"]; got != "degraded_performance" { + t.Fatalf("component checkpoint = %q, want pending degraded status", got) + } + if len(store.pendingComponents) != 0 { + t.Fatalf("pending components = %v, want cleared", store.pendingComponents) + } + + notifier.sends = nil + if err := runner.CheckOnce(context.Background()); err != nil { + t.Fatalf("third CheckOnce returned error: %v", err) + } + if len(notifier.sends) != 2 { + t.Fatalf("third sends = %v, want recovery sent to both subscribers", notifier.sends) + } + if got := store.savedComponents["c1"]; got != "operational" { + t.Fatalf("component checkpoint = %q, want recovery status", got) + } +} + +func summaryWithComponent(id, name, status string) openai.Summary { + return openai.Summary{Components: []openai.Component{{ID: id, Name: name, Status: status, UpdatedAt: "2026-01-01T00:00:00Z"}}} +} diff --git a/internal/redisstore/checkpoint.go b/internal/redisstore/checkpoint.go new file mode 100644 index 0000000..66cac3a --- /dev/null +++ b/internal/redisstore/checkpoint.go @@ -0,0 +1,139 @@ +package redisstore + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +func (s *Store) IsInitialized(ctx context.Context) (bool, error) { + count, err := s.client.Exists(ctx, initializedKey).Result() + return count > 0, err +} + +func (s *Store) SetInitialized(ctx context.Context) error { + return s.client.Set(ctx, initializedKey, "1", 0).Err() +} + +func (s *Store) ComponentStatuses(ctx context.Context) (map[string]string, error) { + return s.client.HGetAll(ctx, componentStatusesKey).Result() +} + +func (s *Store) SaveComponentStatus(ctx context.Context, componentID, status string) error { + return s.client.HSet(ctx, componentStatusesKey, componentID, status).Err() +} + +type PendingComponentEvent struct { + ComponentID string `json:"component_id"` + ComponentName string `json:"component_name"` + Status string `json:"status"` + UpdatedAt string `json:"updated_at"` + Position int `json:"position"` + PreviousStatus string `json:"previous_status"` + DeliveryKey string `json:"delivery_key"` +} + +func (s *Store) PendingComponentEvents(ctx context.Context) (map[string]PendingComponentEvent, error) { + values, err := s.client.HGetAll(ctx, pendingComponentsKey).Result() + if err != nil { + return nil, err + } + events := make(map[string]PendingComponentEvent, len(values)) + for componentID, value := range values { + var event PendingComponentEvent + if err := json.Unmarshal([]byte(value), &event); err != nil { + continue + } + if event.ComponentID == "" { + event.ComponentID = componentID + } + events[componentID] = event + } + return events, nil +} + +func (s *Store) SavePendingComponentEvent(ctx context.Context, event PendingComponentEvent) error { + payload, err := json.Marshal(event) + if err != nil { + return err + } + return s.client.HSet(ctx, pendingComponentsKey, event.ComponentID, payload).Err() +} + +func (s *Store) RemovePendingComponentEvent(ctx context.Context, componentID string) error { + return s.client.HDel(ctx, pendingComponentsKey, componentID).Err() +} + +func (s *Store) HasIncidentUpdate(ctx context.Context, updateID string) (bool, error) { + return s.client.SIsMember(ctx, incidentUpdatesKey, updateID).Result() +} + +func (s *Store) MarkIncidentUpdate(ctx context.Context, updateID string) error { + return s.client.SAdd(ctx, incidentUpdatesKey, updateID).Err() +} + +func (s *Store) HasIncidentUpdateVersion(ctx context.Context, updateID, version string) (bool, error) { + stored, err := s.client.HGet(ctx, incidentVersionsKey, updateID).Result() + if err == nil { + return stored == version, nil + } + if err != redis.Nil { + return false, err + } + return s.client.SIsMember(ctx, incidentUpdatesKey, updateID).Result() +} + +func (s *Store) MarkIncidentUpdateVersion(ctx context.Context, updateID, version string) error { + pipe := s.client.TxPipeline() + pipe.HSet(ctx, incidentVersionsKey, updateID, version) + pipe.SAdd(ctx, incidentUpdatesKey, updateID) + _, err := pipe.Exec(ctx) + return err +} + +func (s *Store) HasDelivered(ctx context.Context, eventKey, subscriberKey string) (bool, error) { + return s.client.SIsMember(ctx, deliveryStateKey(eventKey), subscriberKey).Result() +} + +func (s *Store) MarkDelivered(ctx context.Context, eventKey, subscriberKey string) error { + key := deliveryStateKey(eventKey) + pipe := s.client.TxPipeline() + pipe.SAdd(ctx, key, subscriberKey) + pipe.Expire(ctx, key, 7*24*time.Hour) + _, err := pipe.Exec(ctx) + return err +} + +func (s *Store) ClearDelivery(ctx context.Context, eventKey string) error { + return s.client.Del(ctx, deliveryStateKey(eventKey)).Err() +} + +func (s *Store) TelegramOffset(ctx context.Context) (int64, error) { + value, err := s.client.Get(ctx, telegramOffsetKey).Result() + if err == redis.Nil { + return 0, nil + } + if err != nil { + return 0, err + } + offset, err := strconv.ParseInt(value, 10, 64) + if err != nil { + return 0, fmt.Errorf("invalid telegram offset: %w", err) + } + return offset, nil +} + +func (s *Store) SaveTelegramOffset(ctx context.Context, offset int64) error { + return s.client.Set(ctx, telegramOffsetKey, strconv.FormatInt(offset, 10), 0).Err() +} + +func deliveryStateKey(eventKey string) string { + sum := sha256.Sum256([]byte(eventKey)) + return "openai-status:event-delivery:" + hex.EncodeToString(sum[:]) +} diff --git a/internal/redisstore/store.go b/internal/redisstore/store.go index 97ee595..e9ceed8 100644 --- a/internal/redisstore/store.go +++ b/internal/redisstore/store.go @@ -1,63 +1,23 @@ package redisstore -import ( - "context" - "fmt" - "strconv" - "strings" +import "github.com/redis/go-redis/v9" - "github.com/redis/go-redis/v9" +const ( + subscribersKey = "openai-status:subscribers" + subscriberSettingsKey = "openai-status:subscriber-settings" + componentStatusesKey = "openai-status:component-statuses" + pendingComponentsKey = "openai-status:pending-component-events" + incidentUpdatesKey = "openai-status:incident-updates" + incidentVersionsKey = "openai-status:incident-update-versions" + initializedKey = "openai-status:initialized" + telegramOffsetKey = "openai-status:telegram-offset" ) const ( - subscribersKey = "openai-status:subscribers" - componentStatusesKey = "openai-status:component-statuses" - incidentUpdatesKey = "openai-status:incident-updates" - initializedKey = "openai-status:initialized" + SubscriptionTypeIncident = "incident" + SubscriptionTypeComponent = "component" ) -type Subscriber struct { - ChatID int64 - ThreadID *int -} - -func NewSubscriber(chatID int64, threadID *int) Subscriber { - var copiedThreadID *int - if threadID != nil { - value := *threadID - copiedThreadID = &value - } - return Subscriber{ChatID: chatID, ThreadID: copiedThreadID} -} - -func (s Subscriber) Key() string { - if s.ThreadID == nil { - return strconv.FormatInt(s.ChatID, 10) - } - return fmt.Sprintf("%d:%d", s.ChatID, *s.ThreadID) -} - -func ParseSubscriberKey(value string) (Subscriber, error) { - parts := strings.Split(value, ":") - if len(parts) != 1 && len(parts) != 2 { - return Subscriber{}, fmt.Errorf("invalid subscriber key %q", value) - } - - chatID, err := strconv.ParseInt(parts[0], 10, 64) - if err != nil { - return Subscriber{}, fmt.Errorf("invalid chat ID: %w", err) - } - if len(parts) == 1 { - return Subscriber{ChatID: chatID}, nil - } - - threadID, err := strconv.Atoi(parts[1]) - if err != nil { - return Subscriber{}, fmt.Errorf("invalid thread ID: %w", err) - } - return NewSubscriber(chatID, &threadID), nil -} - type Store struct { client redis.UniversalClient } @@ -65,53 +25,3 @@ type Store struct { func New(client redis.UniversalClient) *Store { return &Store{client: client} } - -func (s *Store) AddSubscriber(ctx context.Context, sub Subscriber) error { - return s.client.SAdd(ctx, subscribersKey, sub.Key()).Err() -} - -func (s *Store) RemoveSubscriber(ctx context.Context, sub Subscriber) error { - return s.client.SRem(ctx, subscribersKey, sub.Key()).Err() -} - -func (s *Store) ListSubscribers(ctx context.Context) ([]Subscriber, error) { - keys, err := s.client.SMembers(ctx, subscribersKey).Result() - if err != nil { - return nil, err - } - - subscribers := make([]Subscriber, 0, len(keys)) - for _, key := range keys { - sub, err := ParseSubscriberKey(key) - if err != nil { - return nil, err - } - subscribers = append(subscribers, sub) - } - return subscribers, nil -} - -func (s *Store) IsInitialized(ctx context.Context) (bool, error) { - count, err := s.client.Exists(ctx, initializedKey).Result() - return count > 0, err -} - -func (s *Store) SetInitialized(ctx context.Context) error { - return s.client.Set(ctx, initializedKey, "1", 0).Err() -} - -func (s *Store) ComponentStatuses(ctx context.Context) (map[string]string, error) { - return s.client.HGetAll(ctx, componentStatusesKey).Result() -} - -func (s *Store) SaveComponentStatus(ctx context.Context, componentID, status string) error { - return s.client.HSet(ctx, componentStatusesKey, componentID, status).Err() -} - -func (s *Store) HasIncidentUpdate(ctx context.Context, updateID string) (bool, error) { - return s.client.SIsMember(ctx, incidentUpdatesKey, updateID).Result() -} - -func (s *Store) MarkIncidentUpdate(ctx context.Context, updateID string) error { - return s.client.SAdd(ctx, incidentUpdatesKey, updateID).Err() -} diff --git a/internal/redisstore/store_test.go b/internal/redisstore/store_test.go index 5c78347..279ec84 100644 --- a/internal/redisstore/store_test.go +++ b/internal/redisstore/store_test.go @@ -30,3 +30,17 @@ func TestParseSubscriberKeyRejectsInvalidValue(t *testing.T) { t.Fatal("expected invalid key error") } } + +func TestSubscriberAcceptsComponentIDAndLegacyNameFilters(t *testing.T) { + sub := NewSubscriber(12345, nil) + sub.Components = []string{"component-id", "Legacy API Name"} + if !sub.Accepts(SubscriptionTypeComponent, "component-id", "API") { + t.Fatal("expected component ID filter to match") + } + if !sub.Accepts(SubscriptionTypeComponent, "other-id", "legacy api name") { + t.Fatal("expected legacy component name filter to match") + } + if sub.Accepts(SubscriptionTypeComponent, "other-id", "Other") { + t.Fatal("unexpected component filter match") + } +} diff --git a/internal/redisstore/subscriber.go b/internal/redisstore/subscriber.go new file mode 100644 index 0000000..5179578 --- /dev/null +++ b/internal/redisstore/subscriber.go @@ -0,0 +1,200 @@ +package redisstore + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "strings" + + "github.com/redis/go-redis/v9" +) + +type Subscriber struct { + ChatID int64 + ThreadID *int + Types []string + Components []string +} + +type subscriberSettings struct { + Types []string `json:"types"` + Components []string `json:"components"` +} + +func NewSubscriber(chatID int64, threadID *int) Subscriber { + var copiedThreadID *int + if threadID != nil { + value := *threadID + copiedThreadID = &value + } + return Subscriber{ + ChatID: chatID, + ThreadID: copiedThreadID, + Types: DefaultSubscriptionTypes(), + Components: []string{}, + } +} + +func (s Subscriber) Key() string { + if s.ThreadID == nil { + return strconv.FormatInt(s.ChatID, 10) + } + return fmt.Sprintf("%d:%d", s.ChatID, *s.ThreadID) +} + +func ParseSubscriberKey(value string) (Subscriber, error) { + parts := strings.Split(value, ":") + if len(parts) != 1 && len(parts) != 2 { + return Subscriber{}, fmt.Errorf("invalid subscriber key %q", value) + } + + chatID, err := strconv.ParseInt(parts[0], 10, 64) + if err != nil { + return Subscriber{}, fmt.Errorf("invalid chat ID: %w", err) + } + if len(parts) == 1 { + return NewSubscriber(chatID, nil), nil + } + + threadID, err := strconv.Atoi(parts[1]) + if err != nil { + return Subscriber{}, fmt.Errorf("invalid thread ID: %w", err) + } + return NewSubscriber(chatID, &threadID), nil +} + +func DefaultSubscriptionTypes() []string { + return []string{SubscriptionTypeIncident, SubscriptionTypeComponent} +} + +func (s Subscriber) Accepts(eventType, componentID, componentName string) bool { + if !containsFold(s.Types, eventType) { + return false + } + if eventType != SubscriptionTypeComponent || len(s.Components) == 0 { + return true + } + return containsFold(s.Components, componentID) || containsFold(s.Components, componentName) +} + +func (s *Store) AddSubscriber(ctx context.Context, sub Subscriber) error { + key := sub.Key() + settings, err := s.loadSubscriberSettings(ctx, key) + if err != nil { + return err + } + if err := s.client.SAdd(ctx, subscribersKey, key).Err(); err != nil { + return err + } + return s.saveSubscriberSettings(ctx, key, settings) +} + +func (s *Store) RemoveSubscriber(ctx context.Context, sub Subscriber) error { + key := sub.Key() + pipe := s.client.TxPipeline() + pipe.SRem(ctx, subscribersKey, key) + pipe.HDel(ctx, subscriberSettingsKey, key) + _, err := pipe.Exec(ctx) + return err +} + +func (s *Store) ListSubscribers(ctx context.Context) ([]Subscriber, error) { + keys, err := s.client.SMembers(ctx, subscribersKey).Result() + if err != nil { + return nil, err + } + + subscribers := make([]Subscriber, 0, len(keys)) + for _, key := range keys { + sub, ok, err := s.loadSubscriber(ctx, key) + if err != nil { + return nil, err + } + if !ok { + continue + } + subscribers = append(subscribers, sub) + } + return subscribers, nil +} + +func (s *Store) GetSubscriber(ctx context.Context, sub Subscriber) (Subscriber, bool, error) { + key := sub.Key() + exists, err := s.client.SIsMember(ctx, subscribersKey, key).Result() + if err != nil || !exists { + return Subscriber{}, false, err + } + return s.loadSubscriber(ctx, key) +} + +func (s *Store) UpdateSubscriberTypes(ctx context.Context, sub Subscriber, types []string) (bool, error) { + key := sub.Key() + settings, exists, err := s.loadExistingSubscriberSettings(ctx, key) + if err != nil || !exists { + return false, err + } + settings.Types = normalizeTypes(types) + return true, s.saveSubscriberSettings(ctx, key, settings) +} + +func (s *Store) UpdateSubscriberComponents(ctx context.Context, sub Subscriber, components []string) (bool, error) { + key := sub.Key() + settings, exists, err := s.loadExistingSubscriberSettings(ctx, key) + if err != nil || !exists { + return false, err + } + settings.Components = normalizeComponents(components) + return true, s.saveSubscriberSettings(ctx, key, settings) +} + +func (s *Store) loadSubscriber(ctx context.Context, key string) (Subscriber, bool, error) { + sub, err := ParseSubscriberKey(key) + if err != nil { + return Subscriber{}, false, nil + } + settings, err := s.loadSubscriberSettings(ctx, key) + if err != nil { + return Subscriber{}, false, err + } + sub.Types = settings.Types + sub.Components = settings.Components + return sub, true, nil +} + +func (s *Store) loadExistingSubscriberSettings(ctx context.Context, key string) (subscriberSettings, bool, error) { + exists, err := s.client.SIsMember(ctx, subscribersKey, key).Result() + if err != nil || !exists { + return subscriberSettings{}, false, err + } + settings, err := s.loadSubscriberSettings(ctx, key) + return settings, true, err +} + +func (s *Store) loadSubscriberSettings(ctx context.Context, key string) (subscriberSettings, error) { + value, err := s.client.HGet(ctx, subscriberSettingsKey, key).Result() + if err == redis.Nil { + return subscriberSettings{Types: DefaultSubscriptionTypes(), Components: []string{}}, nil + } + if err != nil { + return subscriberSettings{}, err + } + + var settings subscriberSettings + if err := json.Unmarshal([]byte(value), &settings); err != nil { + return subscriberSettings{Types: DefaultSubscriptionTypes(), Components: []string{}}, nil + } + settings.Types = normalizeTypes(settings.Types) + settings.Components = normalizeComponents(settings.Components) + return settings, nil +} + +func (s *Store) saveSubscriberSettings(ctx context.Context, key string, settings subscriberSettings) error { + settings.Types = normalizeTypes(settings.Types) + settings.Components = normalizeComponents(settings.Components) + payload, err := json.Marshal(settings) + if err != nil { + return err + } + return s.client.HSet(ctx, subscriberSettingsKey, key, payload).Err() +} diff --git a/internal/redisstore/subscriber_normalization.go b/internal/redisstore/subscriber_normalization.go new file mode 100644 index 0000000..7bc7da9 --- /dev/null +++ b/internal/redisstore/subscriber_normalization.go @@ -0,0 +1,44 @@ +package redisstore + +import "strings" + +func normalizeTypes(types []string) []string { + normalized := make([]string, 0, len(types)) + for _, value := range types { + switch strings.ToLower(strings.TrimSpace(value)) { + case SubscriptionTypeIncident: + if !containsFold(normalized, SubscriptionTypeIncident) { + normalized = append(normalized, SubscriptionTypeIncident) + } + case SubscriptionTypeComponent: + if !containsFold(normalized, SubscriptionTypeComponent) { + normalized = append(normalized, SubscriptionTypeComponent) + } + } + } + if len(normalized) == 0 { + return DefaultSubscriptionTypes() + } + return normalized +} + +func normalizeComponents(components []string) []string { + normalized := make([]string, 0, len(components)) + for _, component := range components { + component = strings.TrimSpace(component) + if component == "" || containsFold(normalized, component) { + continue + } + normalized = append(normalized, component) + } + return normalized +} + +func containsFold(values []string, target string) bool { + for _, value := range values { + if strings.EqualFold(value, target) { + return true + } + } + return false +} diff --git a/internal/telegram/client.go b/internal/telegram/client.go index 625994b..1cfd741 100644 --- a/internal/telegram/client.go +++ b/internal/telegram/client.go @@ -4,8 +4,10 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "net/http" + "strings" "time" "github.com/tiennm99/openai-status-bot/internal/redisstore" @@ -16,6 +18,42 @@ type Client struct { httpClient *http.Client } +type APIError struct { + StatusCode int + Description string +} + +func (e *APIError) Error() string { + return fmt.Sprintf("telegram API %d: %s", e.StatusCode, e.Description) +} + +func IsTerminalSendError(err error) bool { + var apiErr *APIError + if !errors.As(err, &apiErr) { + return false + } + if apiErr.StatusCode == http.StatusForbidden { + return true + } + if apiErr.StatusCode != http.StatusBadRequest { + return false + } + description := strings.ToLower(apiErr.Description) + terminalPhrases := []string{ + "chat not found", + "message thread not found", + "message thread is closed", + "bot was kicked", + "bot is not a member", + } + for _, phrase := range terminalPhrases { + if strings.Contains(description, phrase) { + return true + } + } + return false +} + func NewClient(token string, timeout time.Duration) *Client { return &Client{ baseURL: "https://api.telegram.org/bot" + token, @@ -25,6 +63,20 @@ func NewClient(token string, timeout time.Duration) *Client { } } +func (c *Client) DeleteWebhook(ctx context.Context) error { + payload := map[string]any{"drop_pending_updates": false} + var result json.RawMessage + return c.postJSON(ctx, "/deleteWebhook", payload, &result) +} + +func (c *Client) GetMe(ctx context.Context) (User, error) { + var user User + if err := c.postJSON(ctx, "/getMe", map[string]any{}, &user); err != nil { + return User{}, err + } + return user, nil +} + func (c *Client) GetUpdates(ctx context.Context, offset int64, timeoutSeconds int) ([]Update, error) { payload := map[string]any{ "offset": offset, @@ -45,8 +97,10 @@ func (c *Client) SendMessage(ctx context.Context, sub redisstore.Subscriber, tex func (c *Client) SendText(ctx context.Context, chatID int64, threadID *int, text string) error { payload := map[string]any{ - "chat_id": chatID, - "text": text, + "chat_id": chatID, + "text": text, + "parse_mode": "HTML", + "disable_web_page_preview": true, } if threadID != nil { payload["message_thread_id"] = *threadID @@ -83,7 +137,7 @@ func (c *Client) postJSON(ctx context.Context, path string, payload any, target return fmt.Errorf("decode telegram response: %w", err) } if !envelope.OK { - return fmt.Errorf("telegram API %s: %s", res.Status, envelope.Description) + return &APIError{StatusCode: res.StatusCode, Description: envelope.Description} } if target != nil { if err := json.Unmarshal(envelope.Result, target); err != nil { diff --git a/internal/telegram/client_test.go b/internal/telegram/client_test.go new file mode 100644 index 0000000..6359b77 --- /dev/null +++ b/internal/telegram/client_test.go @@ -0,0 +1,73 @@ +package telegram + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestDeleteWebhookKeepsPendingUpdates(t *testing.T) { + var gotPath string + var payload map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Fatalf("decode payload: %v", err) + } + _, _ = w.Write([]byte(`{"ok":true,"result":true}`)) + })) + defer server.Close() + + client := &Client{baseURL: server.URL, httpClient: server.Client()} + if err := client.DeleteWebhook(context.Background()); err != nil { + t.Fatalf("DeleteWebhook returned error: %v", err) + } + if gotPath != "/deleteWebhook" { + t.Fatalf("path = %s", gotPath) + } + if payload["drop_pending_updates"] != false { + t.Fatalf("drop_pending_updates = %v", payload["drop_pending_updates"]) + } +} + +func TestSendTextUsesHTMLAndDisablesPreview(t *testing.T) { + var payload map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/sendMessage" { + t.Fatalf("unexpected path %s", r.URL.Path) + } + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Fatalf("decode payload: %v", err) + } + _, _ = w.Write([]byte(`{"ok":true,"result":{}}`)) + })) + defer server.Close() + + client := &Client{baseURL: server.URL, httpClient: server.Client()} + if err := client.SendText(context.Background(), 123, nil, "hello"); err != nil { + t.Fatalf("SendText returned error: %v", err) + } + if payload["parse_mode"] != "HTML" { + t.Fatalf("parse_mode = %v", payload["parse_mode"]) + } + if payload["disable_web_page_preview"] != true { + t.Fatalf("disable_web_page_preview = %v", payload["disable_web_page_preview"]) + } +} + +func TestIsTerminalSendError(t *testing.T) { + terminal := &APIError{StatusCode: 403, Description: "Forbidden: bot was blocked by the user"} + if !IsTerminalSendError(terminal) { + t.Fatal("403 blocked should be terminal") + } + chatNotFound := &APIError{StatusCode: 400, Description: "Bad Request: chat not found"} + if !IsTerminalSendError(chatNotFound) { + t.Fatal("chat not found should be terminal") + } + parseError := &APIError{StatusCode: 400, Description: "Bad Request: can't parse entities"} + if IsTerminalSendError(parseError) { + t.Fatal("parse errors should not remove subscribers") + } +} diff --git a/internal/telegram/types.go b/internal/telegram/types.go index 64ea722..85160b1 100644 --- a/internal/telegram/types.go +++ b/internal/telegram/types.go @@ -17,3 +17,10 @@ type Chat struct { Type string `json:"type"` Title string `json:"title,omitempty"` } + +type User struct { + ID int64 `json:"id"` + IsBot bool `json:"is_bot"` + FirstName string `json:"first_name"` + Username string `json:"username"` +}