DO NOT MERGE: Demo mass notification service, created fully using RPC skills

src/sentry/notifications/services/mass_notification/__init__.py

@@ -0,0 +1,2 @@
+from .model import *  # noqa
+from .service import *  # noqa

src/sentry/notifications/services/mass_notification/impl.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+from sentry.integrations.models.integration import Integration
+from sentry.integrations.models.organization_integration import OrganizationIntegration
+from sentry.models.organizationmembermapping import OrganizationMemberMapping
+from sentry.notifications.services.mass_notification.model import RpcMassNotificationResult
+from sentry.notifications.services.mass_notification.service import MassNotificationService
+
+
+class DatabaseBackedMassNotificationService(MassNotificationService):
+    def mass_notify_by_integration(
+        self,
+        *,
+        integration_id: int,
+        message: str,
+    ) -> RpcMassNotificationResult:
+        try:
+            Integration.objects.get(id=integration_id)
+        except Integration.DoesNotExist:
+            return RpcMassNotificationResult(
+                success=False,
+                notified_count=0,
+                error_str=f"Integration {integration_id} not found",
+            )
+
+        org_ids = list(
+            OrganizationIntegration.objects.filter(
+                integration_id=integration_id,
+            ).values_list("organization_id", flat=True)
+        )
+
+        return RpcMassNotificationResult(
+            success=True,
+            notified_count=len(org_ids),
+            organization_ids=org_ids,
+        )
+
+    def mass_notify_by_user_organizations(
+        self,
+        *,
+        user_id: int,
+        message: str,
+    ) -> RpcMassNotificationResult:
+        org_ids = list(
+            OrganizationMemberMapping.objects.filter(
+                user_id=user_id,
+            ).values_list("organization_id", flat=True)
+        )
+
+        if not org_ids:
+            return RpcMassNotificationResult(
+                success=False,
+                notified_count=0,
+                error_str=f"No organizations found for user {user_id}",
+            )
+
+        integration_org_ids = list(
+            OrganizationIntegration.objects.filter(
+                organization_id__in=org_ids,
+            )
+            .values_list("organization_id", flat=True)
+            .distinct()
+        )
+
+        return RpcMassNotificationResult(
+            success=True,
+            notified_count=len(integration_org_ids),
+            organization_ids=integration_org_ids,
+        )
+
+    def mass_notify_by_vibes(
+        self,
+        *,
+        organization_id: int,
+        message: str,
+        vibe: str,
+    ) -> RpcMassNotificationResult:
+        org_integrations = OrganizationIntegration.objects.filter(
+            organization_id=organization_id,
+        ).select_related("integration")
+
+        if not org_integrations.exists():
+            return RpcMassNotificationResult(
+                success=False,
+                notified_count=0,
+                error_str=f"No integrations found for organization {organization_id}",
+            )
+
+        vibe_lower = vibe.lower()
+        matched = [
+            oi
+            for oi in org_integrations
+            if vibe_lower in oi.integration.provider.lower()
+            or vibe_lower in oi.integration.name.lower()
+        ]
+
+        if not matched:
+            matched = list(org_integrations)
+
+        return RpcMassNotificationResult(
+            success=True,
+            notified_count=len(matched),
+            organization_ids=[organization_id],
+        )

src/sentry/notifications/services/mass_notification/model.py

@@ -0,0 +1,15 @@
+# Please do not use
+#     from __future__ import annotations
+# in modules such as this one where hybrid cloud data models or service classes are
+# defined, because we want to reflect on type annotations and avoid forward references.
+
+from pydantic import Field
+
+from sentry.hybridcloud.rpc import RpcModel
+
+
+class RpcMassNotificationResult(RpcModel):
+    success: bool
+    notified_count: int
+    error_str: str | None = None
+    organization_ids: list[int] = Field(default_factory=list)

src/sentry/notifications/services/mass_notification/service.py

@@ -0,0 +1,56 @@
+# Please do not use
+#     from __future__ import annotations
+# in modules such as this one where hybrid cloud data models or service classes are
+# defined, because we want to reflect on type annotations and avoid forward references.
+from abc import abstractmethod
+
+from sentry.hybridcloud.rpc.service import RpcService, rpc_method
+from sentry.notifications.services.mass_notification.model import RpcMassNotificationResult
+from sentry.silo.base import SiloMode
+
+
+class MassNotificationService(RpcService):
+    key = "mass_notification"
+    local_mode = SiloMode.CONTROL
+
+    @classmethod
+    def get_local_implementation(cls) -> RpcService:
+        from sentry.notifications.services.mass_notification.impl import (
+            DatabaseBackedMassNotificationService,
+        )
+
+        return DatabaseBackedMassNotificationService()
+
+    @rpc_method
+    @abstractmethod
+    def mass_notify_by_integration(
+        self,
+        *,
+        integration_id: int,
+        message: str,
+    ) -> RpcMassNotificationResult:
+        pass
+
+    @rpc_method
+    @abstractmethod
+    def mass_notify_by_user_organizations(
+        self,
+        *,
+        user_id: int,
+        message: str,
+    ) -> RpcMassNotificationResult:
+        pass
+
+    @rpc_method
+    @abstractmethod
+    def mass_notify_by_vibes(
+        self,
+        *,
+        organization_id: int,
+        message: str,
+        vibe: str,
+    ) -> RpcMassNotificationResult:
+        pass
+
+
+mass_notification_service = MassNotificationService.create_delegation()

tests/sentry/notifications/services/test_mass_notification.py

@@ -0,0 +1,136 @@
+from sentry.integrations.models.organization_integration import OrganizationIntegration
+from sentry.models.organizationmembermapping import OrganizationMemberMapping
+from sentry.notifications.services.mass_notification.service import mass_notification_service
+from sentry.silo.base import SiloMode
+from sentry.testutils.cases import TransactionTestCase
+from sentry.testutils.silo import all_silo_test, assume_test_silo_mode
+
+
+@all_silo_test
+class MassNotifyByIntegrationTest(TransactionTestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        with assume_test_silo_mode(SiloMode.REGION):
+            self.org1 = self.create_organization()
+            self.org2 = self.create_organization()
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            self.integration = self.create_integration(
+                organization=self.org1,
+                external_id="ext-1",
+                provider="slack",
+            )
+            OrganizationIntegration.objects.create(
+                organization_id=self.org2.id,
+                integration=self.integration,
+            )
+
+    def test_success_multiple_orgs(self) -> None:
+        result = mass_notification_service.mass_notify_by_integration(
+            integration_id=self.integration.id,
+            message="hello",
+        )
+        assert result.success is True
+        assert result.notified_count == 2
+        assert set(result.organization_ids) == {self.org1.id, self.org2.id}
+        assert result.error_str is None
+
+    def test_integration_not_found(self) -> None:
+        result = mass_notification_service.mass_notify_by_integration(
+            integration_id=999999,
+            message="hello",
+        )
+        assert result.success is False
+        assert result.notified_count == 0
+        assert result.error_str is not None
+
+
+@all_silo_test
+class MassNotifyByUserOrganizationsTest(TransactionTestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            self.user = self.create_user()
+        with assume_test_silo_mode(SiloMode.REGION):
+            self.org1 = self.create_organization(owner=self.user)
+            self.org2 = self.create_organization(owner=self.user)
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            self.integration = self.create_integration(
+                organization=self.org1,
+                external_id="ext-1",
+                provider="slack",
+            )
+
+    def test_success(self) -> None:
+        result = mass_notification_service.mass_notify_by_user_organizations(
+            user_id=self.user.id,
+            message="hello",
+        )
+        assert result.success is True
+        assert result.notified_count >= 1
+        assert self.org1.id in result.organization_ids
+
+    def test_no_orgs_user(self) -> None:
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            lonely_user = self.create_user()
+            OrganizationMemberMapping.objects.filter(user_id=lonely_user.id).delete()
+
+        result = mass_notification_service.mass_notify_by_user_organizations(
+            user_id=lonely_user.id,
+            message="hello",
+        )
+        assert result.success is False
+        assert result.notified_count == 0
+        assert result.error_str is not None
+
+
+@all_silo_test
+class MassNotifyByVibesTest(TransactionTestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        with assume_test_silo_mode(SiloMode.REGION):
+            self.org = self.create_organization()
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            self.slack_integration = self.create_integration(
+                organization=self.org,
+                external_id="ext-slack",
+                provider="slack",
+                name="Slack Integration",
+            )
+            self.github_integration = self.create_integration(
+                organization=self.org,
+                external_id="ext-github",
+                provider="github",
+                name="GitHub Integration",
+            )
+
+    def test_vibe_match(self) -> None:
+        result = mass_notification_service.mass_notify_by_vibes(
+            organization_id=self.org.id,
+            message="hello",
+            vibe="slack",
+        )
+        assert result.success is True
+        assert result.notified_count == 1
+        assert self.org.id in result.organization_ids
+
+    def test_no_match_fallback(self) -> None:
+        result = mass_notification_service.mass_notify_by_vibes(
+            organization_id=self.org.id,
+            message="hello",
+            vibe="nonexistent",
+        )
+        assert result.success is True
+        assert result.notified_count == 2
+
+    def test_no_integrations_for_org(self) -> None:
+        with assume_test_silo_mode(SiloMode.REGION):
+            empty_org = self.create_organization()
+
+        result = mass_notification_service.mass_notify_by_vibes(
+            organization_id=empty_org.id,
+            message="hello",
+            vibe="slack",
+        )
+        assert result.success is False
+        assert result.notified_count == 0
+        assert result.error_str is not None