perf(notifications): Index notification settings by recipient

src/sentry/notifications/notificationcontroller.py

@@ -2,6 +2,7 @@
 
 from collections import defaultdict
 from collections.abc import Iterable, Mapping, MutableMapping
+from functools import cached_property
 from typing import Union
 
 from django.db.models import Q
@@ -50,6 +51,39 @@ def sort_settings_by_scope(setting: NotificationSettingOption | NotificationSett
     return 0
 
 
+def _group_by_recipient[T: (NotificationSettingOption, NotificationSettingProvider)](
+    settings: Iterable[T],
+) -> Mapping[tuple[str, int], list[T]]:
+    by_recipient: defaultdict[tuple[str, int], list[T]] = defaultdict(list)
+    for setting in settings:
+        if setting.user_id is not None:
+            by_recipient[("user_id", setting.user_id)].append(setting)
+        if setting.team_id is not None:
+            by_recipient[("team_id", setting.team_id)].append(setting)
+    return by_recipient
+
+
+def _filter_settings[T: (NotificationSettingOption, NotificationSettingProvider)](
+    settings: Iterable[T],
+    by_recipient: Mapping[tuple[str, int], list[T]],
+    kwargs: Mapping[str, object],
+) -> list[T]:
+    # Filters are applied once per recipient, so narrow to that recipient's
+    # settings first instead of scanning every recipient's settings each time.
+    candidates: Iterable[T] = settings
+    for key in ("user_id", "team_id"):
+        recipient_id = kwargs.get(key)
+        if isinstance(recipient_id, int):
+            candidates = by_recipient.get((key, recipient_id), [])
+            break
+
+    return [
+        setting
+        for setting in candidates
+        if all(getattr(setting, arg) == kwargs[arg] for arg in kwargs)
+    ]
+
+
 class NotificationController:
     _setting_options: Iterable[NotificationSettingOption] = []
     _setting_providers: Iterable[NotificationSettingProvider] = []
@@ -154,19 +188,27 @@ def _get_query(self) -> Q:
 
         return project_settings | org_settings | user_settings | team_settings
 
+    @cached_property
+    def _setting_options_by_recipient(
+        self,
+    ) -> Mapping[tuple[str, int], list[NotificationSettingOption]]:
+        return _group_by_recipient(self.get_all_setting_options)
+
+    @cached_property
+    def _setting_providers_by_recipient(
+        self,
+    ) -> Mapping[tuple[str, int], list[NotificationSettingProvider]]:
+        return _group_by_recipient(self.get_all_setting_providers)
+
     def _filter_options(self, **kwargs: object) -> list[NotificationSettingOption]:
-        return [
-            setting
-            for setting in self.get_all_setting_options
-            if all(getattr(setting, arg) == kwargs[arg] for arg in kwargs)
-        ]
+        return _filter_settings(
+            self.get_all_setting_options, self._setting_options_by_recipient, kwargs
+        )
 
     def _filter_providers(self, **kwargs: object) -> list[NotificationSettingProvider]:
-        return [
-            setting
-            for setting in self.get_all_setting_providers
-            if all(getattr(setting, arg) == kwargs[arg] for arg in kwargs)
-        ]
+        return _filter_settings(
+            self.get_all_setting_providers, self._setting_providers_by_recipient, kwargs
+        )
 
     def _get_layered_setting_options(
         self,

tests/sentry/notifications/test_notificationcontroller.py

@@ -214,6 +214,49 @@ def test_filter_setting_providers(self) -> None:
         )
         assert filtered_providers == [self.setting_providers[0]]
 
+    def test_filter_settings_by_recipient(self) -> None:
+        other_user = self.create_user()
+        self.create_member(user=other_user, organization=self.organization, teams=[self.team])
+        other_user_option = add_notification_setting_option(
+            scope_type=NotificationScopeEnum.USER,
+            scope_identifier=other_user.id,
+            type=NotificationSettingEnum.DEPLOY,
+            value=NotificationSettingsOptionEnum.NEVER,
+            user_id=other_user.id,
+        )
+        team_option = add_notification_setting_option(
+            scope_type=NotificationScopeEnum.TEAM,
+            scope_identifier=self.team.id,
+            type=NotificationSettingEnum.DEPLOY,
+            value=NotificationSettingsOptionEnum.ALWAYS,
+            team_id=self.team.id,
+        )
+        team_provider = add_notification_setting_provider(
+            scope_type=NotificationScopeEnum.TEAM,
+            scope_identifier=self.team.id,
+            provider=ExternalProviderEnum.SLACK,
+            type=NotificationSettingEnum.DEPLOY,
+            value=NotificationSettingsOptionEnum.ALWAYS,
+            team_id=self.team.id,
+        )
+        controller = NotificationController(
+            recipients=[self.user, other_user, self.team],
+            project_ids=[self.project.id],
+            organization_id=self.organization.id,
+        )
+
+        assert controller._filter_options(user_id=self.user.id) == self.setting_options
+        assert controller._filter_options(user_id=other_user.id) == [other_user_option]
+        assert controller._filter_options(team_id=self.team.id) == [team_option]
+        assert controller._filter_options(
+            user_id=self.user.id, type=NotificationSettingEnum.DEPLOY.value
+        ) == [self.setting_options[0]]
+        assert controller._filter_options(user_id=self.create_user().id) == []
+
+        assert controller._filter_providers(user_id=self.user.id) == self.setting_providers
+        assert controller._filter_providers(team_id=self.team.id) == [team_provider]
+        assert controller._filter_providers(user_id=other_user.id) == []
+
     def test_layering(self) -> None:
         NotificationSettingOption.objects.all().delete()
         top_level_option = add_notification_setting_option(