diff --git a/internal/redisstore/subscriber.go b/internal/redisstore/subscriber.go index a4f6b94..ab7efa2 100644 --- a/internal/redisstore/subscriber.go +++ b/internal/redisstore/subscriber.go @@ -97,15 +97,30 @@ func (s *Store) ListSubscribers(ctx context.Context) ([]Subscriber, error) { return nil, err } + // Fetch every subscriber's settings in one HGETALL instead of one HGET per + // key. The settings hash mirrors the subscribers set, so its size is the + // subscriber count rather than unbounded history. + settingsByKey, err := s.client.HGetAll(ctx, subscriberSettingsKey).Result() + if err != nil { + return nil, err + } + subscribers := make([]Subscriber, 0, len(keys)) for _, key := range keys { - sub, ok, err := s.loadSubscriber(ctx, key) + sub, err := ParseSubscriberKey(key) + if err != nil { + if removeErr := s.client.SRem(ctx, subscribersKey, key).Err(); removeErr != nil { + return nil, fmt.Errorf("remove malformed subscriber key %q: %w", key, removeErr) + } + return nil, fmt.Errorf("malformed subscriber key %q: %w", key, err) + } + value, present := settingsByKey[key] + settings, err := s.resolveSubscriberSettings(ctx, key, value, present) if err != nil { return nil, err } - if !ok { - continue - } + sub.Types = settings.Types + sub.Components = settings.Components subscribers = append(subscribers, sub) } return subscribers, nil diff --git a/internal/redisstore/subscriber_settings.go b/internal/redisstore/subscriber_settings.go index 4bfffe7..4f27801 100644 --- a/internal/redisstore/subscriber_settings.go +++ b/internal/redisstore/subscriber_settings.go @@ -33,12 +33,23 @@ func (s *Store) loadExistingSubscriberSettings(ctx context.Context, key string) 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 defaultSubscriberSettings(), nil + return s.resolveSubscriberSettings(ctx, key, "", false) } if err != nil { return subscriberSettings{}, err } + return s.resolveSubscriberSettings(ctx, key, value, true) +} +// resolveSubscriberSettings turns a raw settings hash value into decoded +// settings. It lets callers that already hold the value (e.g. a batched +// HGETALL in ListSubscribers) reuse the same decode/default/self-heal logic as +// the single-key HGET path. Absent or corrupt values fall back to defaults, +// dropping the corrupt field so it heals on next write. +func (s *Store) resolveSubscriberSettings(ctx context.Context, key, value string, present bool) (subscriberSettings, error) { + if !present { + return defaultSubscriberSettings(), nil + } settings, err := decodeSubscriberSettings(key, value) if err == nil { return settings, nil