diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..1959a98 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,13 @@ +.git +.gitignore +.env +.env.* +!.env.example +docs/ +plans/ +code-review-report.md +coverage.out +*.test +*.log +openai-status-bot +tmp/ diff --git a/README.md b/README.md index 232d57e..cd68a45 100644 --- a/README.md +++ b/README.md @@ -50,11 +50,11 @@ go run ./cmd/openai-status-bot | `TELEGRAM_BOT_TOKEN` | required | Telegram bot token from BotFather | | `REDIS_ADDR` | `localhost:6379` | Redis address | | `REDIS_PASSWORD` | empty | Redis password | -| `REDIS_DB` | `0` | Redis database number | +| `REDIS_DB` | `0` | Redis database number, valid `0`-`15` | | `OPENAI_STATUS_BASE_URL` | `https://status.openai.com` | OpenAI status page base URL | -| `POLL_INTERVAL` | `1m` | Status check interval | -| `HTTP_TIMEOUT` | `10s` | HTTP request timeout | -| `LOG_LEVEL` | `info` | `debug`, `info`, `warn`, or `error` | +| `POLL_INTERVAL` | `1m` | Status check interval, valid `5s`-`1h` | +| `HTTP_TIMEOUT` | `10s` | HTTP request timeout, valid `1s`-`5m` | +| `LOG_LEVEL` | `info` | `debug`, `info`, `warn`/`warning`, or `error` | ## Notes diff --git a/docs/system-architecture.md b/docs/system-architecture.md index 58cba2c..2cd7804 100644 --- a/docs/system-architecture.md +++ b/docs/system-architecture.md @@ -28,7 +28,7 @@ Redis stores subscribers, subscription settings, polling checkpoints, delivery r | `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 | Legacy seen incident update IDs, retained for migration | +| `openai-status:incident-updates` | set | Legacy seen incident update IDs, retained as write-through metadata; version hash is authoritative | | `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 | @@ -47,5 +47,5 @@ The service uses Telegram `getUpdates`, so it does not need a public webhook URL - 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. +- Malformed subscriber keys are removed from Redis and surfaced as an error for the current poll instead of being skipped silently. - Redis connection failure at startup exits the process. diff --git a/internal/bot/bot.go b/internal/bot/bot.go index 134b3b1..118f172 100644 --- a/internal/bot/bot.go +++ b/internal/bot/bot.go @@ -72,7 +72,11 @@ func (b *Bot) Run(ctx context.Context) error { return nil } b.logger.Warn("get telegram updates", "error", err) - time.Sleep(3 * time.Second) + select { + case <-ctx.Done(): + return nil + case <-time.After(3 * time.Second): + } continue } diff --git a/internal/bot/bot_test.go b/internal/bot/bot_test.go index 674e1ee..df754b5 100644 --- a/internal/bot/bot_test.go +++ b/internal/bot/bot_test.go @@ -2,9 +2,11 @@ package bot import ( "context" + "errors" "log/slog" "strings" "testing" + "time" openai "github.com/tiennm99/openai-status-bot/internal/openai" "github.com/tiennm99/openai-status-bot/internal/redisstore" @@ -12,11 +14,16 @@ import ( ) type fakeTelegramClient struct { - replies []string + replies []string + getUpdatesErr error + onGetUpdates func() } func (f *fakeTelegramClient) GetUpdates(context.Context, int64, int) ([]telegram.Update, error) { - return nil, nil + if f.onGetUpdates != nil { + f.onGetUpdates() + } + return nil, f.getUpdatesErr } func (f *fakeTelegramClient) SendText(_ context.Context, _ int64, _ *int, text string) error { @@ -143,6 +150,21 @@ func TestCommandForOtherBotIsIgnored(t *testing.T) { } } +func TestRunReturnsPromptlyWhenContextCanceledAfterGetUpdatesError(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + tg := &fakeTelegramClient{getUpdatesErr: errors.New("network"), onGetUpdates: cancel} + store := &fakeBotStore{} + bot := New(tg, fakeBotStatusClient{}, store, slog.Default(), "OpenAIStatusBot") + + started := time.Now() + if err := bot.Run(ctx); err != nil { + t.Fatalf("Run returned error: %v", err) + } + if elapsed := time.Since(started); elapsed > 500*time.Millisecond { + t.Fatalf("Run took %s after context cancellation", elapsed) + } +} + func newTestBot(components []openai.Component) (*Bot, *fakeBotStore, *fakeTelegramClient) { tg := &fakeTelegramClient{} store := &fakeBotStore{} diff --git a/internal/config/config.go b/internal/config/config.go index 2a737a3..8916f41 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -3,6 +3,7 @@ package config import ( "fmt" "log/slog" + "net/url" "os" "strconv" "strings" @@ -20,31 +21,46 @@ type Config struct { LogLevel slog.Level } +const ( + minPollInterval = 5 * time.Second + maxPollInterval = time.Hour + minHTTPTimeout = time.Second + maxHTTPTimeout = 5 * time.Minute + minRedisDB = 0 + maxRedisDB = 15 +) + func LoadFromEnv() (Config, error) { cfg := Config{ TelegramBotToken: strings.TrimSpace(os.Getenv("TELEGRAM_BOT_TOKEN")), RedisAddr: getEnv("REDIS_ADDR", "localhost:6379"), - RedisPassword: os.Getenv("REDIS_PASSWORD"), + RedisPassword: strings.TrimSpace(os.Getenv("REDIS_PASSWORD")), OpenAIStatusBaseURL: strings.TrimRight(getEnv("OPENAI_STATUS_BASE_URL", "https://status.openai.com"), "/"), } if cfg.TelegramBotToken == "" { return Config{}, fmt.Errorf("TELEGRAM_BOT_TOKEN is required") } + if err := validateBaseURL("OPENAI_STATUS_BASE_URL", cfg.OpenAIStatusBaseURL); err != nil { + return Config{}, err + } var err error - cfg.RedisDB, err = parseIntEnv("REDIS_DB", 0) + cfg.RedisDB, err = parseBoundedIntEnv("REDIS_DB", 0, minRedisDB, maxRedisDB) if err != nil { return Config{}, err } - cfg.PollInterval, err = parseDurationEnv("POLL_INTERVAL", time.Minute) + cfg.PollInterval, err = parseDurationEnv("POLL_INTERVAL", time.Minute, minPollInterval, maxPollInterval) if err != nil { return Config{}, err } - cfg.HTTPTimeout, err = parseDurationEnv("HTTP_TIMEOUT", 10*time.Second) + cfg.HTTPTimeout, err = parseDurationEnv("HTTP_TIMEOUT", 10*time.Second, minHTTPTimeout, maxHTTPTimeout) + if err != nil { + return Config{}, err + } + cfg.LogLevel, err = parseLogLevel(getEnv("LOG_LEVEL", "info")) if err != nil { return Config{}, err } - cfg.LogLevel = parseLogLevel(getEnv("LOG_LEVEL", "info")) return cfg, nil } @@ -56,7 +72,7 @@ func getEnv(key, fallback string) string { return fallback } -func parseIntEnv(key string, fallback int) (int, error) { +func parseBoundedIntEnv(key string, fallback, minValue, maxValue int) (int, error) { value := strings.TrimSpace(os.Getenv(key)) if value == "" { return fallback, nil @@ -65,10 +81,13 @@ func parseIntEnv(key string, fallback int) (int, error) { if err != nil { return 0, fmt.Errorf("%s must be an integer: %w", key, err) } + if parsed < minValue || parsed > maxValue { + return 0, fmt.Errorf("%s must be between %d and %d", key, minValue, maxValue) + } return parsed, nil } -func parseDurationEnv(key string, fallback time.Duration) (time.Duration, error) { +func parseDurationEnv(key string, fallback, minValue, maxValue time.Duration) (time.Duration, error) { value := strings.TrimSpace(os.Getenv(key)) if value == "" { return fallback, nil @@ -77,21 +96,37 @@ func parseDurationEnv(key string, fallback time.Duration) (time.Duration, error) if err != nil { return 0, fmt.Errorf("%s must be a duration: %w", key, err) } - if parsed <= 0 { - return 0, fmt.Errorf("%s must be positive", key) + if parsed < minValue || parsed > maxValue { + return 0, fmt.Errorf("%s must be between %s and %s", key, minValue, maxValue) } return parsed, nil } -func parseLogLevel(value string) slog.Level { +func validateBaseURL(key, value string) error { + parsed, err := url.Parse(value) + if err != nil { + return fmt.Errorf("%s must be a valid URL: %w", key, err) + } + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return fmt.Errorf("%s must use http or https", key) + } + if parsed.Host == "" { + return fmt.Errorf("%s must include a host", key) + } + return nil +} + +func parseLogLevel(value string) (slog.Level, error) { switch strings.ToLower(strings.TrimSpace(value)) { case "debug": - return slog.LevelDebug + return slog.LevelDebug, nil + case "info": + return slog.LevelInfo, nil case "warn", "warning": - return slog.LevelWarn + return slog.LevelWarn, nil case "error": - return slog.LevelError + return slog.LevelError, nil default: - return slog.LevelInfo + return slog.LevelInfo, fmt.Errorf("LOG_LEVEL must be one of debug, info, warn, warning, or error") } } diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..2323dad --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,84 @@ +package config + +import ( + "log/slog" + "strings" + "testing" + "time" +) + +func TestLoadFromEnvDefaultsAndTrimming(t *testing.T) { + setMinimalEnv(t) + t.Setenv("TELEGRAM_BOT_TOKEN", " token ") + t.Setenv("REDIS_PASSWORD", " redis-pass ") + + cfg, err := LoadFromEnv() + if err != nil { + t.Fatalf("LoadFromEnv returned error: %v", err) + } + if cfg.TelegramBotToken != "token" { + t.Fatalf("TelegramBotToken = %q", cfg.TelegramBotToken) + } + if cfg.RedisPassword != "redis-pass" { + t.Fatalf("RedisPassword = %q", cfg.RedisPassword) + } + if cfg.RedisDB != 0 || cfg.PollInterval != time.Minute || cfg.HTTPTimeout != 10*time.Second { + t.Fatalf("unexpected defaults: %+v", cfg) + } +} + +func TestLoadFromEnvAcceptsWarningLogAlias(t *testing.T) { + setMinimalEnv(t) + t.Setenv("LOG_LEVEL", "warning") + + cfg, err := LoadFromEnv() + if err != nil { + t.Fatalf("LoadFromEnv returned error: %v", err) + } + if cfg.LogLevel != slog.LevelWarn { + t.Fatalf("LogLevel = %v, want warn", cfg.LogLevel) + } +} + +func TestLoadFromEnvRejectsInvalidValues(t *testing.T) { + tests := []struct { + name string + key string + value string + wantErr string + }{ + {name: "redis db below range", key: "REDIS_DB", value: "-1", wantErr: "REDIS_DB must be between 0 and 15"}, + {name: "redis db above range", key: "REDIS_DB", value: "16", wantErr: "REDIS_DB must be between 0 and 15"}, + {name: "poll interval too small", key: "POLL_INTERVAL", value: "1ns", wantErr: "POLL_INTERVAL must be between 5s and 1h0m0s"}, + {name: "http timeout too small", key: "HTTP_TIMEOUT", value: "500ms", wantErr: "HTTP_TIMEOUT must be between 1s and 5m0s"}, + {name: "base url missing scheme", key: "OPENAI_STATUS_BASE_URL", value: "status.openai.com", wantErr: "OPENAI_STATUS_BASE_URL must use http or https"}, + {name: "invalid log level", key: "LOG_LEVEL", value: "verbose", wantErr: "LOG_LEVEL must be one of"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + setMinimalEnv(t) + t.Setenv(tt.key, tt.value) + + _, err := LoadFromEnv() + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("error = %q, want containing %q", err.Error(), tt.wantErr) + } + }) + } +} + +func setMinimalEnv(t *testing.T) { + t.Helper() + t.Setenv("TELEGRAM_BOT_TOKEN", "token") + t.Setenv("REDIS_ADDR", "") + t.Setenv("REDIS_PASSWORD", "") + t.Setenv("REDIS_DB", "") + t.Setenv("OPENAI_STATUS_BASE_URL", "") + t.Setenv("POLL_INTERVAL", "") + t.Setenv("HTTP_TIMEOUT", "") + t.Setenv("LOG_LEVEL", "") +} diff --git a/internal/poller/delivery.go b/internal/poller/delivery.go index feaadf0..1e7d5a7 100644 --- a/internal/poller/delivery.go +++ b/internal/poller/delivery.go @@ -39,7 +39,7 @@ func (r *Runner) notifySubscribers(ctx context.Context, event notificationEvent, } if event.deliveryKey != "" { if err := r.store.MarkDelivered(ctx, event.deliveryKey, subscriberKey); err != nil { - return err + r.logger.Warn("mark telegram delivery", "subscriber", subscriberKey, "event", event.deliveryKey, "error", err) } } } diff --git a/internal/poller/event_collection.go b/internal/poller/event_collection.go index c94ad63..4cf7204 100644 --- a/internal/poller/event_collection.go +++ b/internal/poller/event_collection.go @@ -152,9 +152,11 @@ func (r *Runner) collectIncidentEvents(ctx context.Context, response openai.Inci if err != nil { return nil, nil, err } - checkpoints = append(checkpoints, func(ctx context.Context) error { - return r.store.MarkIncidentUpdateVersion(ctx, update.ID, version) - }) + if !seen { + 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{ @@ -183,7 +185,7 @@ func pendingComponent(event redisstore.PendingComponentEvent) openai.Component { } func IncidentUpdateVersion(update openai.IncidentUpdate) string { - parts := []string{update.UpdatedAt, update.Status, update.Body, update.DisplayAt, update.CreatedAt} + parts := []string{update.Status, update.Body, update.DisplayAt, update.CreatedAt} sum := sha256.Sum256([]byte(strings.Join(parts, "\x00"))) return fmt.Sprintf("%x", sum[:]) } diff --git a/internal/poller/poller_test.go b/internal/poller/poller_test.go index 98c4680..ee9b932 100644 --- a/internal/poller/poller_test.go +++ b/internal/poller/poller_test.go @@ -31,6 +31,7 @@ type fakePollerStore struct { incidentVersions map[string]string subscribers []redisstore.Subscriber delivered map[string]map[string]bool + markDeliveredErr error savedComponents map[string]string markedVersions map[string]string pendingComponents map[string]redisstore.PendingComponentEvent @@ -75,6 +76,9 @@ func (f *fakePollerStore) ListSubscribers(context.Context) ([]redisstore.Subscri } func (f *fakePollerStore) MarkDelivered(_ context.Context, eventKey, subscriberKey string) error { + if f.markDeliveredErr != nil { + return f.markDeliveredErr + } if f.delivered[eventKey] == nil { f.delivered[eventKey] = map[string]bool{} } @@ -183,7 +187,7 @@ func TestCheckOnceRemovesTerminalSubscriberAndCheckpoints(t *testing.T) { 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"}}} + notifier := &fakeNotifier{errors: map[string]error{"2": &telegram.APIError{StatusCode: 200, ErrorCode: 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 { @@ -261,6 +265,56 @@ func TestPendingComponentEventSurvivesReturnToPreviousStatus(t *testing.T) { } } +func TestCheckOnceContinuesWhenMarkDeliveredFailsAfterSend(t *testing.T) { + store := newFakePollerStore() + store.initialized = true + store.componentStatuses["c1"] = "operational" + store.subscribers = []redisstore.Subscriber{redisstore.NewSubscriber(1, nil)} + store.markDeliveredErr = errors.New("redis write failed") + notifier := &fakeNotifier{} + 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(notifier.sends) != 1 || notifier.sends[0] != "1" { + t.Fatalf("sends = %v, want [1]", notifier.sends) + } + if got := store.savedComponents["c1"]; got != "degraded_performance" { + t.Fatalf("component checkpoint = %q", got) + } +} + +func TestIncidentUpdateVersionIgnoresUpdatedAtRefresh(t *testing.T) { + base := openai.IncidentUpdate{ID: "u1", Status: "monitoring", Body: "same", DisplayAt: "2026-01-01T00:00:00Z", CreatedAt: "2026-01-01T00:00:00Z", UpdatedAt: "2026-01-01T00:01:00Z"} + refreshed := base + refreshed.UpdatedAt = "2026-01-01T00:02:00Z" + + if got, want := IncidentUpdateVersion(refreshed), IncidentUpdateVersion(base); got != want { + t.Fatalf("version changed for UpdatedAt-only refresh: got %s, want %s", got, want) + } +} + +func TestCheckOnceDoesNotRemarkSeenIncidentVersion(t *testing.T) { + update := openai.IncidentUpdate{ID: "u1", Body: "same", UpdatedAt: "2026-01-01T00:00:00Z"} + store := newFakePollerStore() + store.initialized = true + store.incidentVersions["u1"] = IncidentUpdateVersion(update) + store.subscribers = []redisstore.Subscriber{redisstore.NewSubscriber(1, nil)} + notifier := &fakeNotifier{} + runner := NewRunner(fakeStatusClient{incidents: openai.IncidentsResponse{Incidents: []openai.Incident{{ID: "i1", Name: "incident", Impact: "minor", IncidentUpdates: []openai.IncidentUpdate{update}}}}}, 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) != 0 { + t.Fatalf("sends = %v, want none", notifier.sends) + } + if len(store.markedVersions) != 0 { + t.Fatalf("markedVersions = %v, want none", store.markedVersions) + } +} + 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 index 66cac3a..060ae24 100644 --- a/internal/redisstore/checkpoint.go +++ b/internal/redisstore/checkpoint.go @@ -46,18 +46,26 @@ func (s *Store) PendingComponentEvents(ctx context.Context) (map[string]PendingC } 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 + event, err := decodePendingComponentEvent(componentID, value) + if err != nil { + return nil, err } events[componentID] = event } return events, nil } +func decodePendingComponentEvent(componentID, value string) (PendingComponentEvent, error) { + var event PendingComponentEvent + if err := json.Unmarshal([]byte(value), &event); err != nil { + return PendingComponentEvent{}, fmt.Errorf("decode pending component event %q: %w", componentID, err) + } + if event.ComponentID == "" { + event.ComponentID = componentID + } + return event, nil +} + func (s *Store) SavePendingComponentEvent(ctx context.Context, event PendingComponentEvent) error { payload, err := json.Marshal(event) if err != nil { @@ -86,7 +94,7 @@ func (s *Store) HasIncidentUpdateVersion(ctx context.Context, updateID, version if err != redis.Nil { return false, err } - return s.client.SIsMember(ctx, incidentUpdatesKey, updateID).Result() + return false, nil } func (s *Store) MarkIncidentUpdateVersion(ctx context.Context, updateID, version string) error { diff --git a/internal/redisstore/store_test.go b/internal/redisstore/store_test.go index 279ec84..5d9e3c5 100644 --- a/internal/redisstore/store_test.go +++ b/internal/redisstore/store_test.go @@ -44,3 +44,38 @@ func TestSubscriberAcceptsComponentIDAndLegacyNameFilters(t *testing.T) { t.Fatal("unexpected component filter match") } } + +func TestDecodePendingComponentEventBackfillsComponentID(t *testing.T) { + event, err := decodePendingComponentEvent("c1", `{"component_name":"API","status":"operational"}`) + if err != nil { + t.Fatalf("decodePendingComponentEvent returned error: %v", err) + } + if event.ComponentID != "c1" { + t.Fatalf("ComponentID = %q, want c1", event.ComponentID) + } +} + +func TestDecodePendingComponentEventRejectsCorruptedPayload(t *testing.T) { + if _, err := decodePendingComponentEvent("c1", `{bad json`); err == nil { + t.Fatal("expected corrupted pending event error") + } +} + +func TestDecodeSubscriberSettingsNormalizesValues(t *testing.T) { + settings, err := decodeSubscriberSettings("123", `{"types":["incident","component","incident"],"components":[" api ","API",""]}`) + if err != nil { + t.Fatalf("decodeSubscriberSettings returned error: %v", err) + } + if len(settings.Types) != 2 || settings.Types[0] != SubscriptionTypeIncident || settings.Types[1] != SubscriptionTypeComponent { + t.Fatalf("Types = %v", settings.Types) + } + if len(settings.Components) != 1 || settings.Components[0] != "api" { + t.Fatalf("Components = %v", settings.Components) + } +} + +func TestDecodeSubscriberSettingsRejectsCorruptedPayload(t *testing.T) { + if _, err := decodeSubscriberSettings("123", `{bad json`); err == nil { + t.Fatal("expected corrupted subscriber settings error") + } +} diff --git a/internal/redisstore/subscriber.go b/internal/redisstore/subscriber.go index 5179578..7c0eb55 100644 --- a/internal/redisstore/subscriber.go +++ b/internal/redisstore/subscriber.go @@ -2,12 +2,9 @@ package redisstore import ( "context" - "encoding/json" "fmt" "strconv" "strings" - - "github.com/redis/go-redis/v9" ) type Subscriber struct { @@ -17,11 +14,6 @@ type Subscriber struct { 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 { @@ -151,7 +143,10 @@ func (s *Store) UpdateSubscriberComponents(ctx context.Context, sub Subscriber, func (s *Store) loadSubscriber(ctx context.Context, key string) (Subscriber, bool, error) { sub, err := ParseSubscriberKey(key) if err != nil { - return Subscriber{}, false, nil + if removeErr := s.client.SRem(ctx, subscribersKey, key).Err(); removeErr != nil { + return Subscriber{}, false, fmt.Errorf("remove malformed subscriber key %q: %w", key, removeErr) + } + return Subscriber{}, false, fmt.Errorf("malformed subscriber key %q: %w", key, err) } settings, err := s.loadSubscriberSettings(ctx, key) if err != nil { @@ -161,40 +156,3 @@ func (s *Store) loadSubscriber(ctx context.Context, key string) (Subscriber, boo 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_settings.go b/internal/redisstore/subscriber_settings.go new file mode 100644 index 0000000..daed4e0 --- /dev/null +++ b/internal/redisstore/subscriber_settings.go @@ -0,0 +1,55 @@ +package redisstore + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/redis/go-redis/v9" +) + +type subscriberSettings struct { + Types []string `json:"types"` + Components []string `json:"components"` +} + +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 + } + + return decodeSubscriberSettings(key, value) +} + +func decodeSubscriberSettings(key, value string) (subscriberSettings, error) { + var settings subscriberSettings + if err := json.Unmarshal([]byte(value), &settings); err != nil { + return subscriberSettings{}, fmt.Errorf("decode subscriber settings %q: %w", key, err) + } + 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/telegram/client.go b/internal/telegram/client.go index 1cfd741..ad31850 100644 --- a/internal/telegram/client.go +++ b/internal/telegram/client.go @@ -14,17 +14,22 @@ import ( ) type Client struct { - baseURL string - httpClient *http.Client + baseURL string + httpClient *http.Client + requestTimeout time.Duration } type APIError struct { StatusCode int + ErrorCode int Description string } func (e *APIError) Error() string { - return fmt.Sprintf("telegram API %d: %s", e.StatusCode, e.Description) + if e.ErrorCode != 0 { + return fmt.Sprintf("telegram API error_code %d (HTTP %d): %s", e.ErrorCode, e.StatusCode, e.Description) + } + return fmt.Sprintf("telegram API HTTP %d: %s", e.StatusCode, e.Description) } func IsTerminalSendError(err error) bool { @@ -32,10 +37,14 @@ func IsTerminalSendError(err error) bool { if !errors.As(err, &apiErr) { return false } - if apiErr.StatusCode == http.StatusForbidden { + code := apiErr.ErrorCode + if code == 0 { + code = apiErr.StatusCode + } + if code == http.StatusForbidden { return true } - if apiErr.StatusCode != http.StatusBadRequest { + if code != http.StatusBadRequest { return false } description := strings.ToLower(apiErr.Description) @@ -56,22 +65,21 @@ func IsTerminalSendError(err error) bool { func NewClient(token string, timeout time.Duration) *Client { return &Client{ - baseURL: "https://api.telegram.org/bot" + token, - httpClient: &http.Client{ - Timeout: timeout + 60*time.Second, - }, + baseURL: "https://api.telegram.org/bot" + token, + httpClient: &http.Client{}, + requestTimeout: timeout, } } 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) + return c.postJSON(ctx, "/deleteWebhook", payload, &result, c.requestTimeout) } func (c *Client) GetMe(ctx context.Context) (User, error) { var user User - if err := c.postJSON(ctx, "/getMe", map[string]any{}, &user); err != nil { + if err := c.postJSON(ctx, "/getMe", map[string]any{}, &user, c.requestTimeout); err != nil { return User{}, err } return user, nil @@ -85,7 +93,7 @@ func (c *Client) GetUpdates(ctx context.Context, offset int64, timeoutSeconds in } var updates []Update - if err := c.postJSON(ctx, "/getUpdates", payload, &updates); err != nil { + if err := c.postJSON(ctx, "/getUpdates", payload, &updates, c.longPollRequestTimeout(timeoutSeconds)); err != nil { return nil, err } return updates, nil @@ -107,22 +115,44 @@ func (c *Client) SendText(ctx context.Context, chatID int64, threadID *int, text } var result json.RawMessage - return c.postJSON(ctx, "/sendMessage", payload, &result) + return c.postJSON(ctx, "/sendMessage", payload, &result, c.requestTimeout) } -func (c *Client) postJSON(ctx context.Context, path string, payload any, target any) error { +func (c *Client) longPollRequestTimeout(timeoutSeconds int) time.Duration { + telegramWait := time.Duration(timeoutSeconds) * time.Second + if telegramWait < 0 { + telegramWait = 0 + } + if c.requestTimeout <= 0 { + return telegramWait + } + return telegramWait + c.requestTimeout +} + +func (c *Client) postJSON(ctx context.Context, path string, payload any, target any, requestTimeout time.Duration) error { body, err := json.Marshal(payload) if err != nil { return err } - req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+path, bytes.NewReader(body)) + requestCtx := ctx + if requestTimeout > 0 { + var cancel context.CancelFunc + requestCtx, cancel = context.WithTimeout(ctx, requestTimeout) + defer cancel() + } + + req, err := http.NewRequestWithContext(requestCtx, http.MethodPost, c.baseURL+path, bytes.NewReader(body)) if err != nil { return err } req.Header.Set("Content-Type", "application/json") - res, err := c.httpClient.Do(req) + httpClient := c.httpClient + if httpClient == nil { + httpClient = http.DefaultClient + } + res, err := httpClient.Do(req) if err != nil { return err } @@ -132,12 +162,13 @@ func (c *Client) postJSON(ctx context.Context, path string, payload any, target OK bool `json:"ok"` Result json.RawMessage `json:"result"` Description string `json:"description"` + ErrorCode int `json:"error_code"` } if err := json.NewDecoder(res.Body).Decode(&envelope); err != nil { return fmt.Errorf("decode telegram response: %w", err) } if !envelope.OK { - return &APIError{StatusCode: res.StatusCode, Description: envelope.Description} + return &APIError{StatusCode: res.StatusCode, ErrorCode: envelope.ErrorCode, 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 index 6359b77..d06b1ec 100644 --- a/internal/telegram/client_test.go +++ b/internal/telegram/client_test.go @@ -3,9 +3,11 @@ package telegram import ( "context" "encoding/json" + "errors" "net/http" "net/http/httptest" "testing" + "time" ) func TestDeleteWebhookKeepsPendingUpdates(t *testing.T) { @@ -58,16 +60,53 @@ func TestSendTextUsesHTMLAndDisablesPreview(t *testing.T) { } func TestIsTerminalSendError(t *testing.T) { - terminal := &APIError{StatusCode: 403, Description: "Forbidden: bot was blocked by the user"} + terminal := &APIError{StatusCode: 200, ErrorCode: 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"} + chatNotFound := &APIError{StatusCode: 200, ErrorCode: 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"} + parseError := &APIError{StatusCode: 200, ErrorCode: 400, Description: "Bad Request: can't parse entities"} if IsTerminalSendError(parseError) { t.Fatal("parse errors should not remove subscribers") } + legacyForbidden := &APIError{StatusCode: 403, Description: "Forbidden"} + if !IsTerminalSendError(legacyForbidden) { + t.Fatal("status-code-only 403 should remain terminal") + } +} + +func TestSendTextCapturesTelegramErrorCode(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"ok":false,"error_code":403,"description":"Forbidden: bot was kicked"}`)) + })) + defer server.Close() + + client := &Client{baseURL: server.URL, httpClient: server.Client()} + err := client.SendText(context.Background(), 123, nil, "hello") + var apiErr *APIError + if !errors.As(err, &apiErr) { + t.Fatalf("SendText error = %v, want APIError", err) + } + if apiErr.StatusCode != http.StatusOK || apiErr.ErrorCode != http.StatusForbidden { + t.Fatalf("APIError = %+v", apiErr) + } + if !IsTerminalSendError(err) { + t.Fatal("telegram error_code 403 should be terminal") + } +} + +func TestNewClientUsesConfiguredTimeoutForOrdinaryRequests(t *testing.T) { + client := NewClient("token", 10*time.Second) + if client.requestTimeout != 10*time.Second { + t.Fatalf("requestTimeout = %s, want 10s", client.requestTimeout) + } + if client.httpClient.Timeout != 0 { + t.Fatalf("httpClient.Timeout = %s, want context-scoped timeout", client.httpClient.Timeout) + } + if got := client.longPollRequestTimeout(50); got != time.Minute { + t.Fatalf("longPollRequestTimeout = %s, want 1m", got) + } } diff --git a/plans/260619-1401-initial-scaffold/plan.md b/plans/260619-1401-initial-scaffold/plan.md deleted file mode 100644 index 842eec7..0000000 --- a/plans/260619-1401-initial-scaffold/plan.md +++ /dev/null @@ -1,27 +0,0 @@ -# Initial Scaffold Plan - -## Status - -Completed. - -## Phases - -| Phase | Status | Output | -|-------|--------|--------| -| Scaffold Go project | Completed | `go.mod`, source layout, Docker files | -| Implement bot runtime | Completed | Telegram long polling, commands | -| Implement status checks | Completed | OpenAI JSON client, one-minute poller | -| Implement Redis storage | Completed | Subscribers and checkpoint keys | -| Verify | Completed | Unit tests and build | - -## Acceptance Criteria - -- Go project exists at `/config/workspace/tiennm99/openai-status-bot`. -- Uses Redis as database for subscribers and poll state. -- Checks OpenAI status every minute by default. -- Telegram users can subscribe and query current status. -- Tests and build pass. - -## Unresolved Questions - -- None.