UN-3494 [FEAT] Email users and groups through PGMQ

backend/adapter_processor_v2/views.py

@@ -58,8 +58,6 @@
 from .models import AdapterInstance, UserDefaultAdapter
 
 notification_plugin = get_plugin("notification")
-if notification_plugin:
-    from plugins.notification.constants import ResourceType
 
 logger = logging.getLogger(__name__)
 
@@ -162,12 +160,11 @@ class AdapterInstanceViewSet(
     def get_notification_resource_type(self, resource: Any) -> str | None:
         if not notification_plugin:
             return None
-        return {
-            "LLM": ResourceType.LLM.value,
-            "EMBEDDING": ResourceType.EMBEDDING.value,
-            "VECTOR_DB": ResourceType.VECTOR_DB.value,
-            "X2TEXT": ResourceType.X2TEXT.value,
-        }.get(resource.adapter_type, ResourceType.LLM.value)
+        from tenant_account_v2.notification_resource_types import (
+            adapter_notification_type,
+        )
+
+        return adapter_notification_type(resource.adapter_type)
 
     def get_permissions(self) -> list[Any]:
         # Frictionless adapters: hidden from non-owners (update/retrieve),
@@ -410,17 +407,6 @@ def destroy(
             raise DeleteAdapterInUseError(adapter_name=adapter_instance.adapter_name)
         return Response(status=status.HTTP_204_NO_CONTENT)
 
-    def partial_update(
-        self, request: Request, *args: tuple[Any], **kwargs: dict[str, Any]
-    ) -> Response:
-        adapter = self.get_object()
-        before = self.snapshot_share_axes(adapter)
-
-        response = super().partial_update(request, *args, **kwargs)
-        if response.status_code == 200 and notification_plugin:
-            self._notify_shared_users(adapter, before, request.data, request.user)
-        return response
-
     @action(detail=True, methods=["post"], url_path="share")
     def share(self, request: Request, pk: str | None = None) -> Response:
         """Apply share state, then clear default-adapter links for any user
@@ -467,42 +453,6 @@ def on_owner_removed(self, resource: AdapterInstance, user: User) -> None:
             return
         self._clear_default_adapter_for_removed_users(resource, {user.pk})
 
-    def _notify_shared_users(
-        self,
-        adapter: AdapterInstance,
-        before: dict[str, set[Any]],
-        request_data: dict[str, Any],
-        actor: Any,
-    ) -> None:
-        """Email users newly added to ``shared_users`` (best-effort)."""
-        users_diff = self.diff_share_axes(adapter, before, request_data).get(
-            "shared_users"
-        )
-        if not (users_diff and users_diff.added):
-            return
-        try:
-            adapter_type_to_resource = {
-                "LLM": ResourceType.LLM.value,
-                "EMBEDDING": ResourceType.EMBEDDING.value,
-                "VECTOR_DB": ResourceType.VECTOR_DB.value,
-                "X2TEXT": ResourceType.X2TEXT.value,
-            }
-            resource_type = adapter_type_to_resource.get(
-                adapter.adapter_type, ResourceType.LLM.value
-            )
-            service_class = notification_plugin["service_class"]
-            notification_service = service_class()
-            notification_service.send_sharing_notification(
-                resource_type=resource_type,
-                resource_name=adapter.adapter_name,
-                resource_id=str(adapter.id),
-                shared_by=actor,
-                shared_to=list(users_diff.added),
-                resource_instance=adapter,
-            )
-        except Exception as e:
-            logger.exception("Failed to send sharing notification: %s", e)
-
     def _clear_default_adapter_for_removed_users(
         self,
         adapter: AdapterInstance,

backend/api_v2/api_deployment_views.py

@@ -466,40 +466,3 @@ def list_of_shared_users(self, request: Request, pk: str | None = None) -> Respo
         instance = self.get_object()
         serializer = SharedUserListSerializer(instance)
         return Response(serializer.data)
-
-    def partial_update(self, request: Request, *args: Any, **kwargs: Any) -> Response:
-        """Override partial_update to handle sharing notifications."""
-        instance = self.get_object()
-        before = self.snapshot_share_axes(instance)
-
-        response = super().partial_update(request, *args, **kwargs)
-        if response.status_code == 200 and notification_plugin:
-            self._notify_shared_users(instance, before, request.data, request.user)
-        return response
-
-    def _notify_shared_users(
-        self,
-        instance: APIDeployment,
-        before: dict[str, set[Any]],
-        request_data: dict[str, Any],
-        actor: Any,
-    ) -> None:
-        """Email users newly added to ``shared_users`` (best-effort)."""
-        users_diff = self.diff_share_axes(instance, before, request_data).get(
-            "shared_users"
-        )
-        if not (users_diff and users_diff.added):
-            return
-        try:
-            service_class = notification_plugin["service_class"]
-            notification_service = service_class()
-            notification_service.send_sharing_notification(
-                resource_type=ResourceType.API_DEPLOYMENT.value,
-                resource_name=instance.display_name,
-                resource_id=str(instance.id),
-                shared_by=actor,
-                shared_to=list(users_diff.added),
-                resource_instance=instance,
-            )
-        except Exception as e:
-            logger.exception("Failed to send sharing notification: %s", e)

backend/backend/internal_base_urls.py

@@ -276,4 +276,10 @@ def test_middleware_debug(request):
         include("prompt_studio.prompt_studio_core_v2.internal_urls"),
         name="prompt_studio_internal",
     ),
+    # Group-sharing email notification APIs
+    path(
+        "v1/group-notification/",
+        include("tenant_account_v2.internal_urls"),
+        name="group_notification_internal",
+    ),
 ]

backend/connector_v2/views.py

@@ -35,7 +35,6 @@
 notification_plugin = get_plugin("notification")
 if notification_plugin:
     from plugins.notification.constants import ResourceType
-    from plugins.notification.sharing_notification import SharingNotificationService
 
 logger = logging.getLogger(__name__)
 
@@ -283,45 +282,3 @@ def perform_destroy(self, instance: ConnectorInstance) -> None:
                 f" named {instance.connector_name}"
             )
             raise DeleteConnectorInUseError(connector_name=instance.connector_name)
-
-    def partial_update(self, request: Request, *args: Any, **kwargs: Any) -> Response:
-        """Override to handle sharing notifications."""
-        instance = self.get_object()
-        before = self.snapshot_share_axes(instance)
-
-        response = super().partial_update(request, *args, **kwargs)
-        if response.status_code == 200 and notification_plugin:
-            self._notify_shared_users(instance, before, request.data, request.user)
-        return response
-
-    def _notify_shared_users(
-        self,
-        instance: ConnectorInstance,
-        before: dict[str, set[Any]],
-        request_data: dict[str, Any],
-        actor: Any,
-    ) -> None:
-        """Email users newly added to ``shared_users`` (best-effort)."""
-        users_diff = self.diff_share_axes(instance, before, request_data).get(
-            "shared_users"
-        )
-        if not (users_diff and users_diff.added):
-            return
-        try:
-            SharingNotificationService().send_sharing_notification(
-                resource_type=ResourceType.CONNECTOR.value,
-                resource_name=instance.connector_name,
-                resource_id=str(instance.id),
-                shared_by=actor,
-                shared_to=list(users_diff.added),
-                resource_instance=instance,
-            )
-            logger.info(
-                "Sent sharing notifications for connector to %d users",
-                len(users_diff.added),
-            )
-        except Exception as e:
-            logger.exception(
-                "Failed to send sharing notification, continuing update though: %s",
-                str(e),
-            )

backend/permissions/membership_views.py

@@ -8,7 +8,7 @@
 from rest_framework.decorators import action
 from rest_framework.request import Request
 from rest_framework.response import Response
-from tenant_account_v2.sharing_helpers import serialize_owner_refs
+from tenant_account_v2.sharing_helpers import retained_user_ids, serialize_owner_refs
 
 from permissions.membership_serializers import AddOwnerSerializer, RemoveOwnerSerializer
 
@@ -86,6 +86,11 @@ def _owner_refs(resource: Any) -> list[dict[str, Any]]:
     # --- notifications: reuse the user-sharing service, best-effort ---
 
     def _notification_context(self, resource: Any) -> tuple[str, str] | None:
+        """``(resource_type, resource_name)``, or ``None`` if not notifiable.
+
+        Also used by ``ResourceShareManagementMixin``, which every host mixes
+        in alongside this one.
+        """
         if not notification_plugin or not self.notification_resource_name_field:
             return None
         resource_type = self.get_notification_resource_type(resource)
@@ -115,6 +120,13 @@ def _notify_owner_removed(self, resource: Any, user: User, actor: Any) -> None:
         ctx = self._notification_context(resource)
         if ctx is None:
             return
+        # The OWNER row is already gone by the time this runs (serializer.save()
+        # ran first) -- but the demoted user may still reach the resource another
+        # way (a group, a direct share, an org-wide share, org admin), same as
+        # the direct-share and group-revoke paths already check.
+        retained = retained_user_ids(resource)
+        if retained is None or user.pk in retained:
+            return
         resource_type, resource_name = ctx
         try:
             notification_plugin["service_class"]().send_access_removed_notification(