fix: address status bot reliability issues

2026-10-14 22:17:28 +00:00 · 2026-06-19 16:52:41 +07:00
1 parent 2a4519106e
commit e10fe0c883
@@ -0,0 +1,13 @@
+.git
+.gitignore
+.env
+.env.*
+!.env.example
+docs/
+plans/
+code-review-report.md
+coverage.out
+*.test
+*.log
+openai-status-bot
+tmp/
@@ -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

@@ -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:<hash>` | 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.
@@ -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
 		}

@@ -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{}
@@ -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")
 	}
 }
@@ -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", "")
+}
@@ -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)
 			}
 		}
 	}
@@ -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[:])
 }
@@ -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"}}}
 }
@@ -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 {
@@ -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")
+	}
+}
@@ -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()
-}
@@ -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()
+}
@@ -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 {
@@ -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)
+	}
 }
@@ -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.