UN-3494 [GATED-FEAT] Email users and groups on access grant and revoke

backend/adapter_processor_v2/views.py

@@ -404,17 +404,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
@@ -461,42 +450,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

@@ -414,40 +414,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

@@ -269,4 +269,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

@@ -36,7 +36,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__)
 
@@ -286,45 +285,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/resource_share_views.py

@@ -1,22 +1,26 @@
 """Shared share-management surface for resource ViewSets.
 
-The mixin is **axis-agnostic** — it operates over the sharing "axes" declared
-in :attr:`ResourceShareManagementMixin.share_axes`. ``shared_users`` is an M2M
-on the resource model, while ``shared_groups`` is stored polymorphically in
-``ResourceGroupShare`` (not an M2M) and routed through the sharing helpers; new
-axes can be added by extending that attribute.
+The mixin is **axis-agnostic** — it reads the sharing "axes" named in
+``_SUPPORTED_SHARE_AXES``. ``shared_users`` is the direct-viewer axis, backed by
+VIEWER membership rows, while ``shared_groups`` is stored polymorphically in
+``ResourceGroupShare`` (not an M2M) and routed through the sharing helpers.
 """
 
-from dataclasses import dataclass, field
-from typing import Any, ClassVar
+import logging
+from typing import Any
 
 from django.db.models import Model
+from plugins import get_plugin
 from rest_framework import status
 from rest_framework.decorators import action
 from rest_framework.exceptions import ValidationError
 from rest_framework.request import Request
 from rest_framework.response import Response
 
+logger = logging.getLogger(__name__)
+
+notification_plugin = get_plugin("notification")
+
 _SUPPORTED_SHARE_AXES = ("shared_users", "shared_groups", "shared_to_org")
 
 
@@ -55,30 +59,78 @@ def _coerce_id_list(axis: str, value: Any) -> list[int]:
     return coerced
 
 
-@dataclass
-class AxisDiff:
-    """Pre/post snapshot for a single share axis (M2M field)."""
-
-    before: set[Any] = field(default_factory=set)
-    after: set[Any] = field(default_factory=set)
-
-    @property
-    def added(self) -> set[Any]:
-        return self.after - self.before
+def _notification_context(view: Any, instance: Any) -> tuple[str, str] | None:
+    """Resolve ``(resource_type, resource_name)`` for the email senders.
 
-    @property
-    def removed(self) -> set[Any]:
-        return self.before - self.after
+    ``None`` when the plugin is absent or the host ViewSet has not opted in by
+    setting ``notification_resource_name_field`` and overriding
+    ``get_notification_resource_type`` (both declared on
+    ``OwnerManagementMixin``, which every share host also mixes in).
+    """
+    name_field = getattr(view, "notification_resource_name_field", None)
+    resolve_type = getattr(view, "get_notification_resource_type", None)
+    if not notification_plugin or not name_field or resolve_type is None:
+        return None
+    resource_type = resolve_type(instance)
+    resource_name = getattr(instance, name_field, None)
+    if resource_type is None or not resource_name:
+        return None
+    return resource_type, resource_name
+
+
+def _users_left_without_access(instance: Model, users: set[Any]) -> list[Any]:
+    """Narrow ``users`` to those with no remaining access to ``instance``.
+
+    Someone dropped from ``shared_users`` may still reach the resource via a
+    group or an org-wide share; telling them their access was removed would be
+    wrong.
+    """
+    if not users:
+        return []
+    from tenant_account_v2.sharing_helpers import compute_effective_members
+
+    retained = {member["user_id"] for member in compute_effective_members(instance)}
+    return [user for user in users if user.pk not in retained]
+
+
+def _send_share_notification(
+    instance: Model, context: tuple[str, str], users: set[Any], actor: Any
+) -> None:
+    """Email users newly granted direct access. Best-effort."""
+    resource_type, resource_name = context
+    try:
+        notification_plugin["service_class"]().send_sharing_notification(
+            resource_type=resource_type,
+            resource_name=resource_name,
+            resource_id=str(instance.pk),
+            shared_by=actor,
+            shared_to=list(users),
+            resource_instance=instance,
+        )
+    except Exception:
+        logger.exception("Failed to send sharing notification for %s", instance.pk)
+
+
+def _send_revoke_notification(
+    instance: Model, context: tuple[str, str], users: list[Any], actor: Any
+) -> None:
+    """Email users whose direct access was revoked. Best-effort."""
+    resource_type, resource_name = context
+    try:
+        notification_plugin["service_class"]().send_access_removed_notification(
+            resource_type=resource_type,
+            resource_name=resource_name,
+            resource_id=str(instance.pk),
+            removed_from=users,
+            removed_by=actor,
+            resource_instance=instance,
+        )
+    except Exception:
+        logger.exception("Failed to send access-removed notification for %s", instance.pk)
 
 
 class ResourceShareManagementMixin:
-    """Adds the shared share-management surface to a resource ViewSet.
-
-    Subclasses declare share axes via :attr:`share_axes`. The default
-    covers ``shared_users`` + ``shared_groups``.
-    """
-
-    share_axes: ClassVar[tuple[str, ...]] = ("shared_users", "shared_groups")
+    """Adds the shared share-management surface to a resource ViewSet."""
 
     @action(detail=True, methods=["post"], url_path="share")
     def share(self, request: Request, pk: str | None = None) -> Response:
@@ -91,15 +143,57 @@ def share(self, request: Request, pk: str | None = None) -> Response:
         users, group-membership for groups) live in
         ``ShareAuthorizationService``.
         """
+        from tenant_account_v2.share_notifications import (
+            notify_resource_group_share_changed,
+        )
         from tenant_account_v2.sharing_helpers import ShareAuthorizationService
 
         resource = self.get_object()  # type: ignore[attr-defined]
         desired = _extract_desired_share_state(request.data)
+        users_before = self._read_axis(resource, "shared_users")
+        groups_before = self._read_axis(resource, "shared_groups")
         ShareAuthorizationService.authorize_and_commit(
             actor=request.user, resource=resource, desired=desired
         )
+        # ``_commit`` is the only atomic block on this path, so it has already
+        # committed — the diffs read persisted state and can never announce a
+        # share that rolled back.
+        resource.refresh_from_db()
+        users_after = self._read_axis(resource, "shared_users")
+        groups_after = self._read_axis(resource, "shared_groups")
+        notify_resource_group_share_changed(
+            resource=resource,
+            added=groups_after - groups_before,
+            removed=groups_before - groups_after,
+            actor=request.user,
+        )
+        self._notify_shared_users(
+            resource, users_after - users_before, users_before - users_after, request.user
+        )
         return Response(status=status.HTTP_200_OK)
 
+    def _notify_shared_users(
+        self,
+        instance: Any,
+        added: set[Any],
+        removed: set[Any],
+        actor: Any,
+        /,
+    ) -> None:
+        """Email users granted or denied direct access.
+
+        Resource type and name come from the host's ``OwnerManagementMixin``
+        seam, so every share host is covered without an override.
+        """
+        context = _notification_context(self, instance)
+        if context is None:
+            return
+        if added:
+            _send_share_notification(instance, context, added, actor)
+        revoked = _users_left_without_access(instance, removed)
+        if revoked:
+            _send_revoke_notification(instance, context, revoked, actor)
+
     @action(detail=True, methods=["get"], url_path="effective-members")
     def effective_members(self, request: Request, pk: str | None = None) -> Response:
         """Return all users with access (direct/group/org), priority-deduped."""
@@ -113,36 +207,6 @@ def effective_members(self, request: Request, pk: str | None = None) -> Response
         members = compute_effective_members(self.get_object())  # type: ignore[attr-defined]
         return Response(EffectiveMemberSerializer(members, many=True).data)
 
-    def snapshot_share_axes(self, instance: Model) -> dict[str, set[Any]]:
-        """Capture every declared axis's current contents.
-
-        Call BEFORE ``super().partial_update(...)``; pair with
-        :meth:`diff_share_axes` afterward.
-        """
-        return {axis: self._read_axis(instance, axis) for axis in self.share_axes}
-
-    def diff_share_axes(
-        self,
-        instance: Model,
-        before: dict[str, set[Any]],
-        request_data: dict[str, Any],
-    ) -> dict[str, AxisDiff]:
-        """Diff each axis that was touched by the request.
-
-        Returns a dict keyed by axis name with only the axes present in
-        ``request_data`` — callers can skip notification fan-out for axes
-        the client did not modify.
-        """
-        instance.refresh_from_db()
-        return {
-            axis: AxisDiff(
-                before=before[axis],
-                after=self._read_axis(instance, axis),
-            )
-            for axis in self.share_axes
-            if axis in request_data
-        }
-
     @staticmethod
     def _read_axis(instance: Model, axis: str) -> set[Any]:
         """Return the current set of related objects on the given axis.