fix(mongo): restore redis parity safeguards

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 {

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
+	}
+}

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)

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
+}