From ea1005c2e84e66a9d148bb86289bfadb90112b4e Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Sat, 27 Jun 2026 16:05:32 +0700 Subject: [PATCH] fix(mongo): restore redis parity safeguards --- .env.example | 2 +- Dockerfile | 2 +- README.md | 2 +- docs/setup-guide.md | 2 +- internal/mongostore/checkpoint.go | 30 ++++- internal/mongostore/indexes.go | 97 +++++++++++--- internal/mongostore/store_test.go | 204 ++++++++++++++++++++++++++++-- internal/mongostore/subscriber.go | 77 +++++++++-- 8 files changed, 377 insertions(+), 39 deletions(-) diff --git a/.env.example b/.env.example index ffcaf02..7c08a23 100644 --- a/.env.example +++ b/.env.example @@ -1,6 +1,6 @@ TELEGRAM_BOT_TOKEN= MONGODB_URI=mongodb+srv://:@/ -MONGODB_DATABASE=development +MONGODB_DATABASE=openai_status_bot POLL_INTERVAL=1m HTTP_TIMEOUT=10s LOG_LEVEL=info diff --git a/Dockerfile b/Dockerfile index 4120c03..0598a10 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM golang:1.24-alpine AS build +FROM golang:1.25-alpine AS build WORKDIR /src COPY go.mod go.sum* ./ diff --git a/README.md b/README.md index 715e75a..16c46a0 100644 --- a/README.md +++ b/README.md @@ -79,7 +79,7 @@ The bot always reads OpenAI status from `https://status.openai.com`. The first successful poll seeds the database and does not send historical incidents. Notifications start from later changes. -Switching from a prior Redis deployment starts from empty state: there is no data migration, so subscribers must re-issue `/start` and component checkpoints reseed on the first poll. +Switching from a prior Redis deployment starts from empty state: there is no data migration, so subscribers must re-issue `/start`, component checkpoints reseed on the first poll, and the stored Telegram update offset is not migrated, so retained Telegram updates can be reprocessed once after cutover. Incident update dedupe tracks the update content/version, so edited Statuspage updates can notify again. Each event is checkpointed independently once it has fully fanned out, so a retryable Telegram failure on one event only defers that event for retry on a later poll and never blocks checkpoints for other events delivered in the same poll. diff --git a/docs/setup-guide.md b/docs/setup-guide.md index 8b3ad77..97792f5 100644 --- a/docs/setup-guide.md +++ b/docs/setup-guide.md @@ -2,7 +2,7 @@ ## Prerequisites -- Go 1.24+ +- Go 1.25+ - MongoDB Atlas cluster (managed, no local MongoDB service required) - Telegram bot token from BotFather diff --git a/internal/mongostore/checkpoint.go b/internal/mongostore/checkpoint.go index a12b617..0d92b3e 100644 --- a/internal/mongostore/checkpoint.go +++ b/internal/mongostore/checkpoint.go @@ -3,6 +3,9 @@ package mongostore import ( "context" "errors" + "fmt" + "strconv" + "strings" "time" "go.mongodb.org/mongo-driver/v2/bson" @@ -166,7 +169,7 @@ func (s *Store) ClearDelivery(ctx context.Context, eventKey string) error { func (s *Store) TelegramOffset(ctx context.Context) (int64, error) { var doc struct { - Value int64 `bson:"value"` + Value any `bson:"value"` } err := s.meta.FindOne(ctx, bson.M{"_id": metaTelegramOffsetID}).Decode(&doc) if errors.Is(err, mongo.ErrNoDocuments) { @@ -175,7 +178,30 @@ func (s *Store) TelegramOffset(ctx context.Context) (int64, error) { if err != nil { return 0, err } - return doc.Value, nil + offset, ok := parseTelegramOffsetValue(doc.Value) + if ok { + return offset, nil + } + if _, err := s.meta.DeleteOne(ctx, bson.M{"_id": metaTelegramOffsetID}); err != nil { + return 0, fmt.Errorf("clear invalid telegram offset: %w", err) + } + return 0, nil +} + +func parseTelegramOffsetValue(value any) (int64, bool) { + switch value := value.(type) { + case int: + return int64(value), true + case int32: + return int64(value), true + case int64: + return value, true + case string: + offset, err := strconv.ParseInt(strings.TrimSpace(value), 10, 64) + return offset, err == nil + default: + return 0, false + } } func (s *Store) SaveTelegramOffset(ctx context.Context, offset int64) error { diff --git a/internal/mongostore/indexes.go b/internal/mongostore/indexes.go index f9702b9..89963b7 100644 --- a/internal/mongostore/indexes.go +++ b/internal/mongostore/indexes.go @@ -8,23 +8,90 @@ import ( "go.mongodb.org/mongo-driver/v2/mongo/options" ) -// deliveryTTLSeconds is the TTL index expiry for delivery markers, in seconds -// (7 days). Mongo's TTL monitor sweeps expired docs roughly once a minute, -// which is irrelevant against a 7-day window. -const deliveryTTLSeconds = int32(deliveryTTL / 1e9) +// deliveryTTLIndexExpireAfterSeconds makes Mongo expire each delivery marker at +// its own expiresAt timestamp. MarkDelivered sets expiresAt to now+7d, matching +// the Redis delivery marker retention window. +const deliveryTTLIndexExpireAfterSeconds = int32(0) -// EnsureIndexes creates the indexes the store relies on. It is idempotent: -// re-creating an existing identical index is a no-op. +// EnsureIndexes creates the indexes the store relies on. It is idempotent, and +// replaces the earlier delivery TTL index shape that expired at expiresAt+7d. func (s *Store) EnsureIndexes(ctx context.Context) error { - _, err := s.delivery.Indexes().CreateMany(ctx, []mongo.IndexModel{ - { - Keys: bson.D{{Key: "expiresAt", Value: 1}}, - Options: options.Index().SetExpireAfterSeconds(deliveryTTLSeconds), - }, - { - // Speeds DeliveredSubscribers/ClearDelivery lookups by eventKey. - Keys: bson.D{{Key: "eventKey", Value: 1}}, - }, + if err := s.ensureDeliveryTTLIndex(ctx); err != nil { + return err + } + _, err := s.delivery.Indexes().CreateOne(ctx, mongo.IndexModel{ + // Speeds DeliveredSubscribers/ClearDelivery lookups by eventKey. + Keys: bson.D{{Key: "eventKey", Value: 1}}, }) return err } + +func (s *Store) ensureDeliveryTTLIndex(ctx context.Context) error { + cursor, err := s.delivery.Indexes().List(ctx) + if err != nil { + return err + } + defer cursor.Close(ctx) + + needsCreate := true + for cursor.Next(ctx) { + var index bson.M + if err := cursor.Decode(&index); err != nil { + return err + } + if !isExpiresAtIndex(index["key"]) { + continue + } + if indexNumber(index["expireAfterSeconds"]) == int64(deliveryTTLIndexExpireAfterSeconds) { + needsCreate = false + continue + } + name, _ := index["name"].(string) + if name == "" { + continue + } + if err := s.delivery.Indexes().DropOne(ctx, name); err != nil { + return err + } + } + if err := cursor.Err(); err != nil { + return err + } + if !needsCreate { + return nil + } + + _, err = s.delivery.Indexes().CreateOne(ctx, mongo.IndexModel{ + Keys: bson.D{{Key: "expiresAt", Value: 1}}, + Options: options.Index().SetExpireAfterSeconds(deliveryTTLIndexExpireAfterSeconds), + }) + return err +} + +func isExpiresAtIndex(key any) bool { + switch key := key.(type) { + case bson.M: + return len(key) == 1 && indexNumber(key["expiresAt"]) == 1 + case map[string]any: + return len(key) == 1 && indexNumber(key["expiresAt"]) == 1 + case bson.D: + return len(key) == 1 && key[0].Key == "expiresAt" && indexNumber(key[0].Value) == 1 + default: + return false + } +} + +func indexNumber(value any) int64 { + switch value := value.(type) { + case int: + return int64(value) + case int32: + return int64(value) + case int64: + return value + case float64: + return int64(value) + default: + return -1 + } +} diff --git a/internal/mongostore/store_test.go b/internal/mongostore/store_test.go index 4244396..afaae32 100644 --- a/internal/mongostore/store_test.go +++ b/internal/mongostore/store_test.go @@ -12,6 +12,7 @@ import ( "strings" "sync" "testing" + "time" "go.mongodb.org/mongo-driver/v2/bson" "go.mongodb.org/mongo-driver/v2/mongo" @@ -29,6 +30,15 @@ var ( // testMongo starts the shared mongod container once, then hands each test a // Store bound to a unique database that is dropped on cleanup. func testMongo(t *testing.T) *Store { + t.Helper() + store := testMongoWithoutIndexes(t) + if err := store.EnsureIndexes(context.Background()); err != nil { + t.Fatalf("EnsureIndexes: %v", err) + } + return store +} + +func testMongoWithoutIndexes(t *testing.T) *Store { t.Helper() ctx := context.Background() @@ -58,11 +68,7 @@ func testMongo(t *testing.T) *Store { _ = db.Drop(context.Background()) }) - store := New(sharedClient, dbName) - if err := store.EnsureIndexes(ctx); err != nil { - t.Fatalf("EnsureIndexes: %v", err) - } - return store + return New(sharedClient, dbName) } func TestAddAndGetSubscriber(t *testing.T) { @@ -111,6 +117,25 @@ func TestAddSubscriberPreservesSettingsOnReStart(t *testing.T) { } } +func TestRemoveSubscriberDeletesDocument(t *testing.T) { + ctx := context.Background() + store := testMongo(t) + sub := NewSubscriber(123, nil) + if err := store.AddSubscriber(ctx, sub); err != nil { + t.Fatalf("AddSubscriber: %v", err) + } + if err := store.RemoveSubscriber(ctx, sub); err != nil { + t.Fatalf("RemoveSubscriber: %v", err) + } + _, exists, err := store.GetSubscriber(ctx, sub) + if err != nil { + t.Fatalf("GetSubscriber after remove: %v", err) + } + if exists { + t.Fatal("subscriber still exists after remove") + } +} + func TestGetSubscriberMissingReturnsFalse(t *testing.T) { ctx := context.Background() store := testMongo(t) @@ -147,6 +172,64 @@ func TestListSubscribersSelfHealsMalformedKey(t *testing.T) { } } +func TestListSubscribersSelfHealsMalformedSettings(t *testing.T) { + ctx := context.Background() + store := testMongo(t) + if _, err := store.subscribers.InsertOne(ctx, bson.M{ + "_id": "2", + "chatID": int64(2), + "types": "not-an-array", + "components": bson.A{123}, + }); err != nil { + t.Fatalf("insert malformed subscriber settings: %v", err) + } + + subs, err := store.ListSubscribers(ctx) + if err != nil { + t.Fatalf("ListSubscribers: %v", err) + } + if len(subs) != 1 || subs[0].Key() != "2" { + t.Fatalf("subscribers = %v, want key 2", subs) + } + if len(subs[0].Types) != 2 || len(subs[0].Components) != 0 { + t.Fatalf("settings = types %v components %v, want defaults", subs[0].Types, subs[0].Components) + } + + var raw bson.M + if err := store.subscribers.FindOne(ctx, bson.M{"_id": "2"}).Decode(&raw); err != nil { + t.Fatalf("load healed subscriber: %v", err) + } + types, ok := storedStringSlice(raw["types"]) + if !ok || len(normalizeTypes(types)) != 2 { + t.Fatalf("healed types = %v ok=%v, want defaults", raw["types"], ok) + } + components, ok := storedStringSlice(raw["components"]) + if !ok || len(components) != 0 { + t.Fatalf("healed components = %v ok=%v, want empty", raw["components"], ok) + } +} + +func TestGetSubscriberSelfHealsMalformedSettings(t *testing.T) { + ctx := context.Background() + store := testMongo(t) + if _, err := store.subscribers.InsertOne(ctx, bson.M{ + "_id": "3", + "chatID": int64(3), + "types": bson.A{"incident"}, + "components": int32(42), + }); err != nil { + t.Fatalf("insert malformed subscriber settings: %v", err) + } + + got, exists, err := store.GetSubscriber(ctx, NewSubscriber(3, nil)) + if err != nil || !exists { + t.Fatalf("GetSubscriber exists=%v err=%v", exists, err) + } + if len(got.Types) != 1 || got.Types[0] != SubscriptionTypeIncident || len(got.Components) != 0 { + t.Fatalf("settings = types %v components %v, want incident and empty components", got.Types, got.Components) + } +} + func TestUpdateSubscriberSettingsMatchedVsMissing(t *testing.T) { ctx := context.Background() store := testMongo(t) @@ -172,6 +255,38 @@ func TestUpdateSubscriberSettingsMatchedVsMissing(t *testing.T) { } } +func TestUpdateSubscriberTypesMatchedVsMissing(t *testing.T) { + ctx := context.Background() + store := testMongo(t) + sub := NewSubscriber(1, nil) + + updated, err := store.UpdateSubscriberTypes(ctx, sub, []string{SubscriptionTypeComponent}) + if err != nil { + t.Fatalf("UpdateSubscriberTypes missing: %v", err) + } + if updated { + t.Fatal("updated = true for missing subscriber, want false") + } + + if err := store.AddSubscriber(ctx, sub); err != nil { + t.Fatalf("AddSubscriber: %v", err) + } + updated, err = store.UpdateSubscriberTypes(ctx, sub, []string{" COMPONENT ", "invalid"}) + if err != nil { + t.Fatalf("UpdateSubscriberTypes existing: %v", err) + } + if !updated { + t.Fatal("updated = false for existing subscriber, want true") + } + got, _, err := store.GetSubscriber(ctx, sub) + if err != nil { + t.Fatalf("GetSubscriber: %v", err) + } + if len(got.Types) != 1 || got.Types[0] != SubscriptionTypeComponent { + t.Fatalf("Types = %v, want normalized [component]", got.Types) + } +} + func TestComponentStatusRoundTrip(t *testing.T) { ctx := context.Background() store := testMongo(t) @@ -260,6 +375,15 @@ func TestDeliveryDedupAndClear(t *testing.T) { if err := store.MarkDelivered(ctx, "evt1", "sub-b"); err != nil { t.Fatalf("MarkDelivered: %v", err) } + var marker struct { + ExpiresAt time.Time `bson:"expiresAt"` + } + if err := store.delivery.FindOne(ctx, bson.M{"_id": "evt1|sub-a"}).Decode(&marker); err != nil { + t.Fatalf("load delivery marker: %v", err) + } + if marker.ExpiresAt.Before(time.Now().Add(deliveryTTL-time.Minute)) || marker.ExpiresAt.After(time.Now().Add(deliveryTTL+time.Minute)) { + t.Fatalf("expiresAt = %s, want about %s from now", marker.ExpiresAt, deliveryTTL) + } delivered, err := store.DeliveredSubscribers(ctx, "evt1") if err != nil { t.Fatalf("DeliveredSubscribers: %v", err) @@ -282,6 +406,26 @@ func TestDeliveryDedupAndClear(t *testing.T) { func TestDeliveryTTLIndexExists(t *testing.T) { ctx := context.Background() store := testMongo(t) + assertDeliveryTTLIndex(t, ctx, store, int64(deliveryTTLIndexExpireAfterSeconds)) +} + +func TestEnsureIndexesReplacesLegacyDeliveryTTLIndex(t *testing.T) { + ctx := context.Background() + store := testMongoWithoutIndexes(t) + if _, err := store.delivery.Indexes().CreateOne(ctx, mongo.IndexModel{ + Keys: bson.D{{Key: "expiresAt", Value: 1}}, + Options: options.Index().SetExpireAfterSeconds(604800), + }); err != nil { + t.Fatalf("create legacy TTL index: %v", err) + } + if err := store.EnsureIndexes(ctx); err != nil { + t.Fatalf("EnsureIndexes: %v", err) + } + assertDeliveryTTLIndex(t, ctx, store, int64(deliveryTTLIndexExpireAfterSeconds)) +} + +func assertDeliveryTTLIndex(t *testing.T, ctx context.Context, store *Store, want int64) { + t.Helper() cursor, err := store.delivery.Indexes().List(ctx) if err != nil { t.Fatalf("list indexes: %v", err) @@ -292,19 +436,22 @@ func TestDeliveryTTLIndexExists(t *testing.T) { } found := false for _, idx := range indexes { - if expire, ok := idx["expireAfterSeconds"]; ok { - if asInt(expire) == 604800 { - found = true - } + if !isExpiresAtIndex(idx["key"]) { + continue + } + if asInt(idx["expireAfterSeconds"]) == want { + found = true } } if !found { - t.Fatalf("no TTL index with expireAfterSeconds=604800 in %v", indexes) + t.Fatalf("no expiresAt TTL index with expireAfterSeconds=%d in %v", want, indexes) } } func asInt(v any) int64 { switch n := v.(type) { + case int: + return int64(n) case int32: return int64(n) case int64: @@ -338,6 +485,43 @@ func TestTelegramOffsetRoundTrip(t *testing.T) { } } +func TestTelegramOffsetAcceptsStringValue(t *testing.T) { + ctx := context.Background() + store := testMongo(t) + if _, err := store.meta.InsertOne(ctx, bson.M{"_id": metaTelegramOffsetID, "value": "42"}); err != nil { + t.Fatalf("insert string offset: %v", err) + } + offset, err := store.TelegramOffset(ctx) + if err != nil { + t.Fatalf("TelegramOffset: %v", err) + } + if offset != 42 { + t.Fatalf("offset = %d, want 42", offset) + } +} + +func TestTelegramOffsetSelfHealsInvalidValue(t *testing.T) { + ctx := context.Background() + store := testMongo(t) + if _, err := store.meta.InsertOne(ctx, bson.M{"_id": metaTelegramOffsetID, "value": "bad"}); err != nil { + t.Fatalf("insert invalid offset: %v", err) + } + offset, err := store.TelegramOffset(ctx) + if err != nil { + t.Fatalf("TelegramOffset: %v", err) + } + if offset != 0 { + t.Fatalf("offset = %d, want 0", offset) + } + count, err := store.meta.CountDocuments(ctx, bson.M{"_id": metaTelegramOffsetID}) + if err != nil { + t.Fatalf("CountDocuments: %v", err) + } + if count != 0 { + t.Fatalf("invalid offset doc count = %d, want 0", count) + } +} + func TestInitializedFlag(t *testing.T) { ctx := context.Background() store := testMongo(t) diff --git a/internal/mongostore/subscriber.go b/internal/mongostore/subscriber.go index 38bbd2f..866d20b 100644 --- a/internal/mongostore/subscriber.go +++ b/internal/mongostore/subscriber.go @@ -23,11 +23,11 @@ type Subscriber struct { // subscriber key), mirroring the Redis design where the key held identity and a // separate hash held settings; chatID/threadID are denormalized for inspection. type subscriberDoc struct { - ID string `bson:"_id"` - ChatID int64 `bson:"chatID"` - ThreadID *int `bson:"threadID,omitempty"` - Types []string `bson:"types"` - Components []string `bson:"components"` + ID string `bson:"_id"` + ChatID int64 `bson:"chatID"` + ThreadID *int `bson:"threadID,omitempty"` + Types any `bson:"types"` + Components any `bson:"components"` } func NewSubscriber(chatID int64, threadID *int) Subscriber { @@ -180,7 +180,9 @@ func (s *Store) setSubscriberFields(ctx context.Context, key string, fields bson } // subscriberFromDoc derives identity from the document _id (so a malformed key -// self-heals as it did under Redis) and normalizes the stored settings. +// self-heals as it did under Redis) and normalizes the stored settings. Corrupt +// or missing settings fields are rewritten to defaults so one malformed +// document cannot block subscriber fan-out. func (s *Store) subscriberFromDoc(ctx context.Context, doc subscriberDoc) (Subscriber, error) { sub, err := ParseSubscriberKey(doc.ID) if err != nil { @@ -189,7 +191,66 @@ func (s *Store) subscriberFromDoc(ctx context.Context, doc subscriberDoc) (Subsc } return Subscriber{}, fmt.Errorf("malformed subscriber key %q: %w", doc.ID, err) } - sub.Types = normalizeTypes(doc.Types) - sub.Components = normalizeComponents(doc.Components) + + types, healTypes := normalizeStoredTypes(doc.Types) + components, healComponents := normalizeStoredComponents(doc.Components) + if healTypes || healComponents { + fields := bson.M{} + if healTypes { + fields["types"] = types + } + if healComponents { + fields["components"] = components + } + if _, err := s.subscribers.UpdateOne(ctx, bson.M{"_id": doc.ID}, bson.M{"$set": fields}); err != nil { + return Subscriber{}, fmt.Errorf("heal malformed subscriber settings %q: %w", doc.ID, err) + } + } + + sub.Types = types + sub.Components = components return sub, nil } + +func normalizeStoredTypes(value any) ([]string, bool) { + values, ok := storedStringSlice(value) + if !ok { + return DefaultSubscriptionTypes(), true + } + return normalizeTypes(values), false +} + +func normalizeStoredComponents(value any) ([]string, bool) { + values, ok := storedStringSlice(value) + if !ok { + return []string{}, true + } + return normalizeComponents(values), false +} + +func storedStringSlice(value any) ([]string, bool) { + switch value := value.(type) { + case nil: + return nil, false + case []string: + return append([]string(nil), value...), true + case bson.A: + return stringsFromAnySlice([]any(value)) + case []any: + return stringsFromAnySlice(value) + default: + return nil, false + } +} + +func stringsFromAnySlice(values []any) ([]string, bool) { + result := make([]string, 0, len(values)) + for _, value := range values { + text, ok := value.(string) + if !ok { + return nil, false + } + result = append(result, text) + } + return result, true +}