fix(hc): Introduce Organization.get_bulk_owner_profiles

src/sentry/models/organization.py

@@ -37,7 +37,7 @@
 from sentry.roles.manager import Role
 from sentry.services.hybrid_cloud.notifications import notifications_service
 from sentry.services.hybrid_cloud.organization_mapping import organization_mapping_service
-from sentry.services.hybrid_cloud.user import RpcUser
+from sentry.services.hybrid_cloud.user import RpcUser, RpcUserContact
 from sentry.services.hybrid_cloud.user.service import user_service
 from sentry.types.organization import OrganizationAbsoluteUrlMixin
 from sentry.utils.http import is_using_customer_domain
@@ -329,6 +329,48 @@ def default_owner_id(self):
             self._default_owner_id = owners[0].id
         return self._default_owner_id
 
+    @classmethod
+    def _get_bulk_owner_ids(cls, organizations: Collection[Organization]) -> dict[int, int]:
+        from sentry.models.organizationmember import OrganizationMember
+
+        owner_id_table: dict[int, int] = {}
+        org_ids_to_query: list[int] = []
+        for org in organizations:
+            default_owner = getattr(org, "_default_owner", None)
+            if default_owner:
+                owner_id_table[org.id] = default_owner.id
+            else:
+                org_ids_to_query.append(org.id)
+
+        if org_ids_to_query:
+            queried_owner_ids = OrganizationMember.objects.filter(
+                organization_id__in=org_ids_to_query, role=roles.get_top_dog().id
+            ).values_list("organization_id", "user_id")
+
+            for (org_id, user_id) in queried_owner_ids:
+                # An org may have multiple owners. Here we mimic the behavior of
+                # `get_default_owner`, which is to use the first one in the query
+                # result's iteration order.
+                if org_id not in owner_id_table:
+                    owner_id_table[org_id] = user_id
+
+        return owner_id_table
+
+    @classmethod
+    def get_bulk_owner_contacts(
+        cls, organizations: Collection[Organization]
+    ) -> dict[int, RpcUserContact]:
+        owner_id_table = cls._get_bulk_owner_ids(organizations)
+
+        contacts = user_service.get_bulk_contact_info(user_ids=list(owner_id_table.values()))
+        contact_table = {c.id: c for c in contacts}
+
+        return {
+            org_id: contact_table[user_id]
+            for (org_id, user_id) in owner_id_table.items()
+            if user_id in contact_table
+        }
+
     def has_single_owner(self):
         owners = list(
             self.get_members_with_org_roles([roles.get_top_dog().id])[:2].values_list(

src/sentry/services/hybrid_cloud/user/impl.py

@@ -37,7 +37,11 @@
     UserSerializeType,
     UserUpdateArgs,
 )
-from sentry.services.hybrid_cloud.user.model import RpcVerifyUserEmail, UserIdEmailArgs
+from sentry.services.hybrid_cloud.user.model import (
+    RpcUserContact,
+    RpcVerifyUserEmail,
+    UserIdEmailArgs,
+)
 from sentry.services.hybrid_cloud.user.serial import serialize_rpc_user, serialize_user_avatar
 from sentry.services.hybrid_cloud.user.service import UserService
 from sentry.signals import user_signup
@@ -283,6 +287,10 @@ def get_user_avatar(self, *, user_id: int) -> RpcAvatar | None:
 
         return serialize_user_avatar(avatar=possible_avatar)
 
+    def get_bulk_contact_info(self, *, user_ids: list[int]) -> list[RpcUserContact]:
+        users = User.objects.filter(id__in=user_ids).values_list("id", "name", "email")
+        return [RpcUserContact(id=id, name=name, email=email) for (id, name, email) in users]
+
     class _UserFilterQuery(
         FilterQueryDatabaseImpl[User, UserFilterArgs, RpcUser, UserSerializeType],
     ):

src/sentry/services/hybrid_cloud/user/model.py

@@ -25,6 +25,12 @@ class RpcUserEmail(RpcModel):
     is_verified: bool = False
 
 
+class RpcUserContact(RpcModel):
+    id: int = -1
+    name: str = ""
+    email: str = ""
+
+
 class RpcAuthenticator(RpcModel):
     id: int = 0
     user_id: int = -1

src/sentry/services/hybrid_cloud/user/service.py

@@ -17,7 +17,12 @@
     UserSerializeType,
     UserUpdateArgs,
 )
-from sentry.services.hybrid_cloud.user.model import RpcAvatar, RpcVerifyUserEmail, UserIdEmailArgs
+from sentry.services.hybrid_cloud.user.model import (
+    RpcAvatar,
+    RpcUserContact,
+    RpcVerifyUserEmail,
+    UserIdEmailArgs,
+)
 from sentry.silo import SiloMode
 
 
@@ -195,6 +200,11 @@ def verify_user_emails(
     def get_user_avatar(self, *, user_id: int) -> RpcAvatar | None:
         pass
 
+    @rpc_method
+    @abstractmethod
+    def get_bulk_contact_info(self, *, user_ids: list[int]) -> list[RpcUserContact]:
+        pass
+
 
 @back_with_silo_cache("user_service.get_user", SiloMode.REGION, RpcUser)
 def get_user(user_id: int) -> RpcUser | None: