openai-status-bot/internal/mongostore/subscriber.go

package mongostore

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"

	"go.mongodb.org/mongo-driver/v2/bson"
	"go.mongodb.org/mongo-driver/v2/mongo"
	"go.mongodb.org/mongo-driver/v2/mongo/options"
)

type Subscriber struct {
	ChatID     int64
	ThreadID   *int
	Types      []string
	Components []string
}

// subscriberDoc is the stored shape of a subscriber. Identity lives in _id (the
// 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      any    `bson:"types"`
	Components any    `bson:"components"`
}

func NewSubscriber(chatID int64, threadID *int) Subscriber {
	var copiedThreadID *int
	if threadID != nil {
		value := *threadID
		copiedThreadID = &value
	}
	return Subscriber{
		ChatID:     chatID,
		ThreadID:   copiedThreadID,
		Types:      DefaultSubscriptionTypes(),
		Components: []string{},
	}
}

func (s Subscriber) Key() string {
	if s.ThreadID == nil {
		return strconv.FormatInt(s.ChatID, 10)
	}
	return fmt.Sprintf("%d:%d", s.ChatID, *s.ThreadID)
}

func ParseSubscriberKey(value string) (Subscriber, error) {
	parts := strings.Split(value, ":")
	if len(parts) != 1 && len(parts) != 2 {
		return Subscriber{}, fmt.Errorf("invalid subscriber key %q", value)
	}

	chatID, err := strconv.ParseInt(parts[0], 10, 64)
	if err != nil {
		return Subscriber{}, fmt.Errorf("invalid chat ID: %w", err)
	}
	if len(parts) == 1 {
		return NewSubscriber(chatID, nil), nil
	}

	threadID, err := strconv.Atoi(parts[1])
	if err != nil {
		return Subscriber{}, fmt.Errorf("invalid thread ID: %w", err)
	}
	return NewSubscriber(chatID, &threadID), nil
}

func DefaultSubscriptionTypes() []string {
	return []string{SubscriptionTypeIncident, SubscriptionTypeComponent}
}

func (s Subscriber) Accepts(eventType, componentID, componentName string) bool {
	if !containsFold(s.Types, eventType) {
		return false
	}
	if eventType != SubscriptionTypeComponent || len(s.Components) == 0 {
		return true
	}
	return containsFold(s.Components, componentID) || containsFold(s.Components, componentName)
}

// AddSubscriber upserts the subscriber. An existing subscriber keeps its
// settings on re-/start (only chatID/threadID are refreshed); a new subscriber
// is seeded with default types and no component filter.
func (s *Store) AddSubscriber(ctx context.Context, sub Subscriber) error {
	// Identity lives in _id; chatID/threadID are denormalized for inspection.
	// Omit threadID when absent so the stored shape matches subscriberDoc's
	// omitempty decode (no explicit null).
	set := bson.M{"chatID": sub.ChatID}
	if sub.ThreadID != nil {
		set["threadID"] = *sub.ThreadID
	}
	update := bson.M{
		"$set": set,
		"$setOnInsert": bson.M{
			"types":      DefaultSubscriptionTypes(),
			"components": []string{},
		},
	}
	_, err := s.subscribers.UpdateOne(ctx, bson.M{"_id": sub.Key()}, update, options.UpdateOne().SetUpsert(true))
	return err
}

func (s *Store) RemoveSubscriber(ctx context.Context, sub Subscriber) error {
	_, err := s.subscribers.DeleteOne(ctx, bson.M{"_id": sub.Key()})
	return err
}

func (s *Store) ListSubscribers(ctx context.Context) ([]Subscriber, error) {
	cursor, err := s.subscribers.Find(ctx, bson.M{})
	if err != nil {
		return nil, err
	}
	defer cursor.Close(ctx)

	var subscribers []Subscriber
	for cursor.Next(ctx) {
		var doc subscriberDoc
		if err := cursor.Decode(&doc); err != nil {
			return nil, err
		}
		sub, err := s.subscriberFromDoc(ctx, doc)
		if err != nil {
			return nil, err
		}
		subscribers = append(subscribers, sub)
	}
	if err := cursor.Err(); err != nil {
		return nil, err
	}
	return subscribers, nil
}

func (s *Store) GetSubscriber(ctx context.Context, sub Subscriber) (Subscriber, bool, error) {
	var doc subscriberDoc
	if err := s.subscribers.FindOne(ctx, bson.M{"_id": sub.Key()}).Decode(&doc); err != nil {
		if errors.Is(err, mongo.ErrNoDocuments) {
			return Subscriber{}, false, nil
		}
		return Subscriber{}, false, err
	}
	resolved, err := s.subscriberFromDoc(ctx, doc)
	if err != nil {
		return Subscriber{}, false, err
	}
	return resolved, true, nil
}

func (s *Store) UpdateSubscriberTypes(ctx context.Context, sub Subscriber, types []string) (bool, error) {
	return s.setSubscriberFields(ctx, sub.Key(), bson.M{"types": normalizeTypes(types)})
}

func (s *Store) UpdateSubscriberComponents(ctx context.Context, sub Subscriber, components []string) (bool, error) {
	return s.setSubscriberFields(ctx, sub.Key(), bson.M{"components": normalizeComponents(components)})
}

func (s *Store) UpdateSubscriberSettings(ctx context.Context, sub Subscriber, types, components []string) (bool, error) {
	return s.setSubscriberFields(ctx, sub.Key(), bson.M{
		"types":      normalizeTypes(types),
		"components": normalizeComponents(components),
	})
}

// setSubscriberFields applies a $set only when the subscriber exists.
// MatchedCount==0 means "not subscribed", replacing the Redis check-then-set
// Lua script.
func (s *Store) setSubscriberFields(ctx context.Context, key string, fields bson.M) (bool, error) {
	result, err := s.subscribers.UpdateOne(ctx, bson.M{"_id": key}, bson.M{"$set": fields})
	if err != nil {
		return false, err
	}
	return result.MatchedCount > 0, nil
}

// subscriberFromDoc derives identity from the document _id (so a malformed key
// 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 {
		if _, removeErr := s.subscribers.DeleteOne(ctx, bson.M{"_id": doc.ID}); removeErr != nil {
			return Subscriber{}, fmt.Errorf("remove malformed subscriber key %q: %w", doc.ID, removeErr)
		}
		return Subscriber{}, fmt.Errorf("malformed subscriber key %q: %w", doc.ID, err)
	}

	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
}