diff --git a/backend/adapter_processor_v2/views.py b/backend/adapter_processor_v2/views.py
index 5d08fc6877..5e638ba8a9 100644
--- a/backend/adapter_processor_v2/views.py
+++ b/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,
diff --git a/backend/api_v2/api_deployment_views.py b/backend/api_v2/api_deployment_views.py
index 891ed1c570..842bce8416 100644
--- a/backend/api_v2/api_deployment_views.py
+++ b/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)
diff --git a/backend/backend/internal_base_urls.py b/backend/backend/internal_base_urls.py
index 30d4c8df22..c7022affb5 100644
--- a/backend/backend/internal_base_urls.py
+++ b/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",
+ ),
]
diff --git a/backend/connector_v2/views.py b/backend/connector_v2/views.py
index 28f8038213..ac60db475b 100644
--- a/backend/connector_v2/views.py
+++ b/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),
- )
diff --git a/backend/permissions/membership_views.py b/backend/permissions/membership_views.py
index e9ade97e9c..6c296beb30 100644
--- a/backend/permissions/membership_views.py
+++ b/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(
diff --git a/backend/permissions/resource_share_views.py b/backend/permissions/resource_share_views.py
index a38ff400f1..a042f5d782 100644
--- a/backend/permissions/resource_share_views.py
+++ b/backend/permissions/resource_share_views.py
@@ -1,22 +1,28 @@
"""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 write side accepts exactly the three axes named in
+``_SUPPORTED_SHARE_AXES``; the read side (``_read_axis``) only knows the two
+per-recipient ones by name. ``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.
"""
+import logging
from dataclasses import dataclass, field
from typing import Any, ClassVar
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,9 +61,71 @@ def _coerce_id_list(axis: str, value: Any) -> list[int]:
return coerced
+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 retained_user_ids
+
+ retained = retained_user_ids(instance)
+ if retained is None:
+ # Org-wide or frictionless: access never depended on the share, so
+ # nobody lost anything.
+ return []
+ 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.
+
+ Sent inline, matching every other direct-share call site in the codebase
+ (pipelines, API deployments, connectors, ...). Group shares, by contrast,
+ dispatch through PGMQ (see ``share_notifications.notify_resource_group_share_changed``)
+ because a group's member count -- and so its send time -- is unbounded in
+ a way a handful of direct users is not.
+ """
+ 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)
+
+
@dataclass
class AxisDiff:
- """Pre/post snapshot for a single share axis (M2M field)."""
+ """DEPRECATED shim -- see ``ResourceShareManagementMixin.share_axes`` below."""
before: set[Any] = field(default_factory=set)
after: set[Any] = field(default_factory=set)
@@ -72,14 +140,36 @@ def removed(self) -> set[Any]:
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``.
- """
+ """Adds the shared share-management surface to a resource ViewSet."""
+ # DEPRECATED, temporary: ``share_axes``, ``snapshot_share_axes`` and
+ # ``diff_share_axes`` (with ``AxisDiff`` above) were removed on this branch
+ # -- the PATCH-based sharing path they backed is dead, its diff is always
+ # empty (see 999e443b4). Restored here only because cloud's
+ # ``AgenticProjectViewSet.partial_update`` on ``origin/main`` still calls
+ # them, and OSS merges before cloud (Zipstack/unstract-cloud#1698 carries
+ # the real removal). Delete this whole block, and this comment, once #1698
+ # merges -- at that point nothing on cloud main calls it anymore.
share_axes: ClassVar[tuple[str, ...]] = ("shared_users", "shared_groups")
+ def snapshot_share_axes(self, instance: Model) -> dict[str, set[Any]]:
+ """DEPRECATED shim. See the block comment above ``share_axes``."""
+ 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]:
+ """DEPRECATED shim. See the block comment above ``share_axes``."""
+ 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
+ }
+
@action(detail=True, methods=["post"], url_path="share")
def share(self, request: Request, pk: str | None = None) -> Response:
"""Apply a replace-style share state for the resource.
@@ -90,15 +180,87 @@ 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)
+ # Only read an axis this request actually touches. authorize_and_commit
+ # takes long enough (auth checks, a DB write) that a concurrent request
+ # changing an axis this one left alone would otherwise land inside the
+ # window and get diffed as if this request made the change — the wrong
+ # actor's name in the notification.
+ users_before = (
+ self._read_axis(resource, "shared_users")
+ if "shared_users" in desired
+ else set()
+ )
+ groups_before = (
+ self._read_axis(resource, "shared_groups")
+ if "shared_groups" in desired
+ else set()
+ )
ShareAuthorizationService.authorize_and_commit(
actor=request.user, resource=resource, desired=desired
)
+ # ``authorize_and_commit`` has already committed here, on the current
+ # deployment: ``ATOMIC_REQUESTS`` is a settings knob, currently off, so
+ # this view isn't wrapped in a transaction and the diffs below read
+ # persisted state. Flipping that knob would flip this premise too.
+ resource.refresh_from_db()
+ # Only the two per-recipient axes notify. ``shared_to_org`` is left out
+ # deliberately: a toggle has no recipient list short of the whole org,
+ # and it is read below as a reason someone KEPT access, not lost it.
+ users_after = (
+ self._read_axis(resource, "shared_users")
+ if "shared_users" in desired
+ else users_before
+ )
+ groups_after = (
+ self._read_axis(resource, "shared_groups")
+ if "shared_groups" in desired
+ else groups_before
+ )
+ 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. Best-effort.
+
+ Resource type and name come from the host's ``OwnerManagementMixin``
+ seam. The share has already committed by the time this runs, so no
+ failure here — a raising seam, a dropped DB connection — may surface
+ as a 500 on a share that succeeded.
+ """
+ try:
+ context = self._notification_context(instance) # type: ignore[attr-defined]
+ 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)
+ except Exception:
+ logger.exception("Failed to send share notifications for %s", instance.pk)
+
@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."""
@@ -112,36 +274,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.
diff --git a/backend/permissions/tests/test_owner_management.py b/backend/permissions/tests/test_owner_management.py
index 1b1b124a9c..b5804799f7 100644
--- a/backend/permissions/tests/test_owner_management.py
+++ b/backend/permissions/tests/test_owner_management.py
@@ -16,7 +16,7 @@
import pytest
from account_v2.models import User
from django.test import TestCase
-from permissions.roles import ResourceRole
+from prompt_studio.permission import ParentToolAccess
from rest_framework import status
from rest_framework.parsers import JSONParser
from rest_framework.request import Request as DRFRequest
@@ -27,7 +27,7 @@
from workflow_manager.workflow_v2.views import WorkflowViewSet
from permissions.membership_serializers import AddOwnerSerializer
-from prompt_studio.permission import ParentToolAccess
+from permissions.roles import ResourceRole
from permissions.tests.base import (
RESOURCE_SPECS,
CoOwnerOrgTestMixin,
@@ -126,7 +126,8 @@ def test_can_remove_when_multiple_owners(self) -> None:
def test_remove_rejects_service_account(self) -> None:
"""Symmetric with the add-side guard: a service-account owner cannot be
- removed, so it can't be stranded off the resource (UN-2202 review #7)."""
+ removed, so it can't be stranded off the resource (UN-2202 review #7).
+ """
svc = make_user("svc@example.com", is_service_account=True)
self.workflow.memberships.create(user=svc, role=ResourceRole.OWNER)
response = self._remove(self.owner, svc.pk)
@@ -136,7 +137,8 @@ def test_remove_rejects_service_account(self) -> None:
def test_shared_viewer_can_list_shared_users(self) -> None:
"""``list_of_shared_users`` is viewer-tier by design (spec §10): a shared
user may open the owner popup. Guards against a re-tighten to IsOwner
- (UN-2202 review #6)."""
+ (UN-2202 review #6).
+ """
self.workflow.memberships.create(user=self.viewer, role=ResourceRole.VIEWER)
view = WorkflowViewSet.as_view({"get": "list_of_shared_users"})
request = self.factory.get("/x/")
@@ -193,7 +195,8 @@ def test_has_members_mixin_accessors(self) -> None:
def test_service_account_excluded_from_owner_roster(self) -> None:
"""A service-account OWNER row is a real grant but must not surface as a
- removable co-owner or inflate the count badge (UN-2202 review #7)."""
+ removable co-owner or inflate the count badge (UN-2202 review #7).
+ """
svc = make_user("svc@example.com", is_service_account=True)
self.workflow.memberships.create(user=self.coowner, role=ResourceRole.OWNER)
self.workflow.memberships.create(user=svc, role=ResourceRole.OWNER)
@@ -203,9 +206,7 @@ def test_service_account_excluded_from_owner_roster(self) -> None:
self.assertEqual(self.workflow.co_owners_count(), 2)
# the filter is display-only — the SA still genuinely owns the resource
self.assertTrue(
- self.workflow.memberships.filter(
- user=svc, role=ResourceRole.OWNER
- ).exists()
+ self.workflow.memberships.filter(user=svc, role=ResourceRole.OWNER).exists()
)
def test_membership_save_derives_organization(self) -> None:
@@ -218,7 +219,8 @@ def test_membership_save_derives_organization(self) -> None:
class CrossResourceOwnerManagementTests(CoOwnerOrgTestMixin, TestCase):
"""The shared owner-management surface behaves identically for every OSS
- shareable resource (AgenticProject is covered cloud-side)."""
+ shareable resource (AgenticProject is covered cloud-side).
+ """
def setUp(self) -> None:
self._seed_org()
@@ -255,7 +257,8 @@ class OwnerNotificationWiringTests(CoOwnerOrgTestMixin, TestCase):
notifications with the right payload, and swallow notification failures so
the owners request is never broken (best-effort). The resource-type hook is
patched to a fixed value so the test does not depend on the cloud-only
- notification plugin's conditional ``ResourceType`` import."""
+ notification plugin's conditional ``ResourceType`` import.
+ """
def setUp(self) -> None:
self._seed_org()
@@ -313,6 +316,20 @@ def test_remove_fires_removed_notification_with_payload(self) -> None:
self.assertEqual(kwargs["resource_id"], str(self.workflow.pk))
self.assertEqual(kwargs["resource_instance"], self.workflow)
+ def test_remove_skips_notification_when_demoted_owner_retains_access(self) -> None:
+ """Demoted from OWNER, but an org admin still reaches every resource --
+ nothing was actually taken from them, so no "access removed" email.
+ """
+ self.workflow.memberships.create(user=self.admin, role=ResourceRole.OWNER)
+ with patch(
+ "account_v2.authentication_controller.AuthenticationController"
+ ".is_admin_by_role",
+ side_effect=lambda role: role == "admin",
+ ):
+ response = self._remove(self.owner, self.admin.pk)
+ self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
+ self.service.send_access_removed_notification.assert_not_called()
+
def test_notification_failure_does_not_break_add(self) -> None:
self.service.send_co_owner_added_notification.side_effect = RuntimeError("boom")
response = self._add(self.owner, self.coowner.pk)
@@ -369,9 +386,7 @@ def test_parent_tool_viewer_allowed_outsider_denied(self) -> None:
def test_null_parent_falls_back_to_object_creator(self) -> None:
# No parent tool → access derives from the object's own ``created_by``.
- orphan = SimpleNamespace(
- prompt_studio_tool=None, created_by_id=self.owner.pk
- )
+ orphan = SimpleNamespace(prompt_studio_tool=None, created_by_id=self.owner.pk)
self.assertTrue(self._perm(self.owner, orphan))
self.assertFalse(self._perm(self.coowner, orphan))
@@ -425,9 +440,7 @@ def test_owner_exempt_from_default_adapter_clear(self) -> None:
before = {self.owner.pk, self.coowner.pk, self.viewer.pk}
after = {self.owner.pk}
with (
- patch.object(
- AdapterInstanceViewSet, "get_object", return_value=self.adapter
- ),
+ patch.object(AdapterInstanceViewSet, "get_object", return_value=self.adapter),
patch.object(
AdapterInstanceViewSet,
"_effective_member_ids",
@@ -566,9 +579,7 @@ def test_adapter_create_grants_creator_ownership(self) -> None:
from adapter_processor_v2.views import AdapterInstanceViewSet
# Only the SDK context-window lookup (a provider-shaped call) is mocked.
- with patch.object(
- AdapterInstance, "get_context_window_size", return_value=4096
- ):
+ with patch.object(AdapterInstance, "get_context_window_size", return_value=4096):
response = self._create(
AdapterInstanceViewSet,
{
@@ -645,7 +656,8 @@ def test_api_deployment_create_grants_creator_ownership(self) -> None:
def test_import_path_grants_creator_ownership(self) -> None:
"""The 7th grant site — ``create_tool_from_import_data`` — is a helper
- the viewset sweep can't reach; pin it directly."""
+ the viewset sweep can't reach; pin it directly.
+ """
from prompt_studio.prompt_studio_core_v2.models import CustomTool
from prompt_studio.prompt_studio_core_v2.prompt_studio_helper import (
PromptStudioHelper,
@@ -659,3 +671,35 @@ def test_import_path_grants_creator_ownership(self) -> None:
)
self.assertTrue(tool.is_owner(self.coowner))
self.assertIn(tool, CustomTool.objects.for_user(self.coowner))
+
+
+class DeprecatedAxisShimTests(CoOwnerOrgTestMixin, TestCase):
+ """The temporary ``share_axes``/``snapshot_share_axes``/``diff_share_axes``
+ shim (see the block comment in ``resource_share_views.py``) -- restored
+ only so cloud's still-stale ``origin/main`` ``AgenticProjectViewSet``
+ doesn't ``AttributeError`` once this branch merges and deletes the real
+ thing on OSS main. Delete this test alongside that block, once
+ Zipstack/unstract-cloud#1698 merges.
+ """
+
+ def setUp(self) -> None:
+ self._seed_org()
+ self.workflow = Workflow.objects.create(
+ workflow_name="wf-shim", organization=self.org, created_by=self.owner
+ )
+ self.workflow.memberships.create(user=self.owner, role=ResourceRole.OWNER)
+
+ def test_snapshot_then_diff_reproduces_the_original_always_empty_shape(self) -> None:
+ # Matches the pre-removal contract exactly: the PATCH path this shim
+ # exists for has no way to actually change shared_users/shared_groups
+ # (see 999e443b4), so the diff for either axis is always empty.
+ view = WorkflowViewSet()
+ before = view.snapshot_share_axes(self.workflow)
+ self.assertEqual(before, {"shared_users": set(), "shared_groups": set()})
+ diff = view.diff_share_axes(
+ self.workflow, before, {"shared_users": [], "shared_groups": []}
+ )
+ self.assertEqual(set(diff), {"shared_users", "shared_groups"})
+ for axis_diff in diff.values():
+ self.assertEqual(axis_diff.added, set())
+ self.assertEqual(axis_diff.removed, set())
diff --git a/backend/permissions/tests/test_share_notifications.py b/backend/permissions/tests/test_share_notifications.py
new file mode 100644
index 0000000000..3c7b7f1794
--- /dev/null
+++ b/backend/permissions/tests/test_share_notifications.py
@@ -0,0 +1,104 @@
+"""Integration tests for direct-user share/revoke email wiring (UN-3494).
+
+The ``share/`` endpoint's ``shared_users`` axis mails users who gained or lost
+direct access. Both notification seams are mocked, so these pin the wiring and
+the payload — who is mailed, with what, and that a failing send never breaks a
+share that already committed — not template or transport behavior. The group
+axis is covered by ``ResourceShareNotificationTests`` in
+``tenant_account_v2.tests``.
+
+DB-backed (Django ``TestCase``), so ``backend/conftest.py`` auto-marks these
+``integration`` and the rig runs them in ``integration-backend``.
+"""
+
+from unittest.mock import Mock, patch
+
+from account_v2.models import User
+from django.test import TestCase
+from rest_framework import status
+from rest_framework.response import Response
+from rest_framework.test import APIRequestFactory, force_authenticate
+from workflow_manager.workflow_v2.models.workflow import Workflow
+from workflow_manager.workflow_v2.views import WorkflowViewSet
+
+from permissions.roles import ResourceRole
+from permissions.tests.base import CoOwnerOrgTestMixin
+
+
+class DirectShareNotificationWiringTests(CoOwnerOrgTestMixin, TestCase):
+ """``POST share/`` mails users whose direct access was granted or revoked."""
+
+ def setUp(self) -> None:
+ self._seed_org()
+ self.workflow = Workflow.objects.create(
+ workflow_name="wf-1", organization=self.org, created_by=self.owner
+ )
+ self.workflow.memberships.create(user=self.owner, role=ResourceRole.OWNER)
+ self.factory = APIRequestFactory()
+ self.service = Mock()
+ plugin = {"service_class": Mock(return_value=self.service)}
+ for p in (
+ # The sender lives in the share mixin; ``_notification_context``
+ # gates on the membership_views copy, so both need the plugin.
+ patch("permissions.resource_share_views.notification_plugin", plugin),
+ patch("permissions.membership_views.notification_plugin", plugin),
+ patch.object(
+ WorkflowViewSet,
+ "get_notification_resource_type",
+ return_value="workflow",
+ ),
+ ):
+ p.start()
+ self.addCleanup(p.stop)
+
+ def _share(self, actor: User, payload: dict) -> Response:
+ view = WorkflowViewSet.as_view({"post": "share"})
+ request = self.factory.post("/x/", payload, format="json")
+ force_authenticate(request, user=actor)
+ return view(request, pk=str(self.workflow.pk))
+
+ def test_granting_direct_access_fires_sharing_notification(self) -> None:
+ response = self._share(self.owner, {"shared_users": [self.viewer.pk]})
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.service.send_sharing_notification.assert_called_once()
+ kwargs = self.service.send_sharing_notification.call_args.kwargs
+ self.assertEqual(kwargs["resource_type"], "workflow")
+ self.assertEqual(kwargs["resource_name"], "wf-1")
+ self.assertEqual(kwargs["resource_id"], str(self.workflow.pk))
+ self.assertEqual(kwargs["shared_by"], self.owner)
+ self.assertEqual([u.pk for u in kwargs["shared_to"]], [self.viewer.pk])
+ self.assertEqual(kwargs["resource_instance"], self.workflow)
+ self.service.send_access_removed_notification.assert_not_called()
+
+ def test_revoking_direct_access_fires_access_removed_notification(self) -> None:
+ self.workflow.memberships.create(user=self.viewer, role=ResourceRole.VIEWER)
+ response = self._share(self.owner, {"shared_users": []})
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.service.send_access_removed_notification.assert_called_once()
+ kwargs = self.service.send_access_removed_notification.call_args.kwargs
+ self.assertEqual(kwargs["resource_type"], "workflow")
+ self.assertEqual([u.pk for u in kwargs["removed_from"]], [self.viewer.pk])
+ self.assertEqual(kwargs["removed_by"], self.owner)
+ self.assertEqual(kwargs["resource_id"], str(self.workflow.pk))
+ self.service.send_sharing_notification.assert_not_called()
+
+ def test_revoke_is_silent_when_the_user_keeps_access_another_way(self) -> None:
+ # Dropped from ``shared_users`` but still covered by the org-wide share —
+ # nothing was lost, so telling them it was removed would be wrong.
+ self.workflow.memberships.create(user=self.viewer, role=ResourceRole.VIEWER)
+ response = self._share(self.owner, {"shared_users": [], "shared_to_org": True})
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.service.send_access_removed_notification.assert_not_called()
+
+ def test_notification_failure_does_not_break_the_share(self) -> None:
+ # The share commits before the mail goes out; a raising sender must not
+ # surface as a 500 on a share that succeeded.
+ self.service.send_sharing_notification.side_effect = RuntimeError("boom")
+ response = self._share(self.owner, {"shared_users": [self.viewer.pk]})
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ viewer_ids = set(
+ self.workflow.memberships.filter(role=ResourceRole.VIEWER).values_list(
+ "user_id", flat=True
+ )
+ )
+ self.assertIn(self.viewer.pk, viewer_ids)
diff --git a/backend/pipeline_v2/views.py b/backend/pipeline_v2/views.py
index 27c800518f..6ac0597e0c 100644
--- a/backend/pipeline_v2/views.py
+++ b/backend/pipeline_v2/views.py
@@ -42,8 +42,6 @@
from pipeline_v2.serializers.sharing import SharedUserListSerializer
notification_plugin = get_plugin("notification")
-if notification_plugin:
- from plugins.notification.constants import ResourceType
logger = logging.getLogger(__name__)
@@ -61,12 +59,13 @@ class PipelineViewSet(
notification_resource_name_field = "pipeline_name"
def get_notification_resource_type(self, resource: Any) -> str | None:
- # Only ETL/TASK pipelines map to a notification ResourceType.
if not notification_plugin:
return None
- if resource.pipeline_type in (ResourceType.ETL.value, ResourceType.TASK.value):
- return resource.pipeline_type
- return None
+ from tenant_account_v2.notification_resource_types import (
+ pipeline_notification_type,
+ )
+
+ return pipeline_notification_type(resource.pipeline_type)
def get_permissions(self) -> list[Any]:
# Enabling or disabling is use, not configuration, so it follows
@@ -194,60 +193,6 @@ def list_of_shared_users(self, request: Request, pk: str | None = None) -> Respo
serializer = SharedUserListSerializer(pipeline)
return Response(serializer.data, status=status.HTTP_200_OK)
- 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: Pipeline,
- before: dict[str, set[Any]],
- request_data: dict[str, Any],
- actor: Any,
- ) -> None:
- """Email users newly added to ``shared_users`` (best-effort).
-
- Only ETL/TASK pipelines map to a notification ``ResourceType``;
- DEFAULT/APP pipelines have no analogue and skip the fan-out.
- """
- users_diff = self.diff_share_axes(instance, before, request_data).get(
- "shared_users"
- )
- if not (users_diff and users_diff.added):
- return
- if instance.pipeline_type not in (
- ResourceType.ETL.value,
- ResourceType.TASK.value,
- ):
- return
- try:
- service_class = notification_plugin["service_class"]
- notification_service = service_class()
- notification_service.send_sharing_notification(
- resource_type=instance.pipeline_type,
- resource_name=instance.pipeline_name,
- resource_id=str(instance.id),
- shared_by=actor,
- shared_to=list(users_diff.added),
- resource_instance=instance,
- )
- logger.info(
- "Sent sharing notifications for %s to %d users",
- instance.pipeline_type,
- len(users_diff.added),
- )
- except Exception as e:
- logger.exception(
- "Failed to send sharing notification, continuing update though: %s",
- str(e),
- )
-
@action(detail=True, methods=["get"])
def download_postman_collection(
self, request: Request, pk: str | None = None
diff --git a/backend/prompt_studio/prompt_studio_core_v2/views.py b/backend/prompt_studio/prompt_studio_core_v2/views.py
index 6da770a060..2c638c87ae 100644
--- a/backend/prompt_studio/prompt_studio_core_v2/views.py
+++ b/backend/prompt_studio/prompt_studio_core_v2/views.py
@@ -280,54 +280,6 @@ def destroy(
)
return super().destroy(request, *args, **kwargs)
- def partial_update(
- self, request: Request, *args: tuple[Any], **kwargs: dict[str, Any]
- ) -> Response:
- custom_tool = self.get_object()
- before = self.snapshot_share_axes(custom_tool)
-
- response = super().partial_update(request, *args, **kwargs)
- if response.status_code == 200:
- self._notify_shared_users(custom_tool, before, request.data, request.user)
- return response
-
- def _notify_shared_users(
- self,
- custom_tool: CustomTool,
- before: dict[str, set[Any]],
- request_data: dict[str, Any],
- actor: Any,
- ) -> None:
- """Email users newly added to ``shared_users`` (best-effort)."""
- notification_plugin = get_plugin("notification")
- if not notification_plugin:
- return
- users_diff = self.diff_share_axes(custom_tool, before, request_data).get(
- "shared_users"
- )
- if not (users_diff and users_diff.added):
- return
-
- from plugins.notification.constants import ResourceType
-
- try:
- service_class = notification_plugin["service_class"]
- notification_service = service_class()
- notification_service.send_sharing_notification(
- resource_type=ResourceType.TEXT_EXTRACTOR.value,
- resource_name=custom_tool.tool_name,
- resource_id=str(custom_tool.tool_id),
- shared_by=actor,
- shared_to=list(users_diff.added),
- resource_instance=custom_tool,
- )
- except Exception as e:
- logger.exception(
- "Failed to send sharing notification for custom tool %s: %s",
- custom_tool.tool_id,
- str(e),
- )
-
@action(detail=True, methods=["get"])
def get_select_choices(self, request: HttpRequest) -> Response:
"""Method to return all static dropdown field values.
diff --git a/backend/tenant_account_v2/group_notification_service.py b/backend/tenant_account_v2/group_notification_service.py
new file mode 100644
index 0000000000..c95929ca07
--- /dev/null
+++ b/backend/tenant_account_v2/group_notification_service.py
@@ -0,0 +1,502 @@
+"""Send-side logic for group-sharing email notifications.
+
+Reached over the internal API by the notification worker. The enqueue side
+(:mod:`tenant_account_v2.share_notifications`) only records *what happened*;
+everything that needs Django — group expansion, org re-validation, resource
+lookup, the email plugin — happens here, because ``workers/`` has no Django.
+
+Sending is a cloud plugin. In OSS ``notification_plugin`` is empty and every
+entry point below no-ops cleanly.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+from account_v2.models import Organization, User
+from django.apps import apps
+from django.conf import settings
+from django.db.models import QuerySet
+from plugins import get_plugin
+
+from tenant_account_v2.models import (
+ GroupMembership,
+ OrganizationGroup,
+ OrganizationMember,
+)
+from tenant_account_v2.notification_resource_types import (
+ adapter_notification_type,
+ pipeline_notification_type,
+)
+from tenant_account_v2.share_notifications import MembershipAction, ShareAction
+from tenant_account_v2.shareable_resources import ShareableResource, descriptor_for_kind
+
+if TYPE_CHECKING:
+ from collections.abc import Iterable
+ from datetime import datetime
+
+logger = logging.getLogger(__name__)
+
+notification_plugin = get_plugin("notification")
+
+# Ceiling on concurrent per-group sends, mirroring the email plugin's own
+# ``MAX_CONCURRENT_CHUNK_SENDS`` -- a resource shared with many groups would
+# otherwise serialize one blocking SendGrid call per group.
+_MAX_CONCURRENT_GROUP_SENDS = 10
+
+# OSS ``ShareableResource.kind`` → the email plugin's ``ResourceType`` value,
+# for the 6 kinds that are a plain 1:1 rename. Pipelines and adapters resolve
+# via ``notification_resource_types`` instead, the same helpers the
+# direct-share viewsets use, so a new adapter/pipeline type only needs
+# registering once. These 6 are still a second, hand-maintained copy of what
+# each ViewSet's own ``get_notification_resource_type`` already states --
+# unifying them the same way is a larger change than this one, tracked
+# separately. Plain strings here, not the cloud enum: this dict is only ever
+# read once the plugin is confirmed loaded (see ``_service()``), but nothing
+# enforces that path if a future caller reached it another way.
+_STATIC_RESOURCE_TYPES = {
+ "workflow": "workflow",
+ "api_deployment": "api",
+ "connector_instance": "connector",
+ "custom_tool": "text_extractor",
+ "agentic_project": "agentic_project",
+ "lookup": "lookup",
+}
+
+
+class ResourceNotFoundError(Exception):
+ """The shared resource no longer exists, or is not in the given org."""
+
+
+@dataclass(frozen=True)
+class _SharedResource:
+ """A resolved resource, reused across every group email in one task."""
+
+ instance: Any
+ name: str
+ type: str | None
+
+
+def send_resource_shared(
+ *,
+ organization: Organization,
+ group_ids: Iterable[int],
+ actor_id: int,
+ resource_kind: str,
+ resource_id: str,
+ share_action: str,
+ revoked_at: datetime | None = None,
+) -> bool:
+ """Mail every current member of each group whose resource access changed.
+
+ One email per group, so ``group_name`` in the template is always the group
+ the recipient actually belongs to. ``share_action`` picks the wording, and
+ on a revoke ``revoked_at`` bounds who counts as "current".
+
+ Returns:
+ ``False`` only when a group with real recipients was actually attempted
+ and the plugin reported a send failure -- the caller's cue to ask for
+ redelivery. Every other outcome (nothing to send, plugin absent,
+ misconfigured) is ``True``: there is nothing a retry would fix.
+ """
+ service = _service()
+ if service is None:
+ return True
+ resolved = _resolve_share(organization, actor_id, resource_kind, resource_id)
+ if resolved is None:
+ return True
+ actor, shared = resolved
+ retained = _retained_user_ids(organization, shared.instance, share_action)
+ if retained is None:
+ return True
+ groups = list(_groups_to_mail(organization, group_ids, shared.instance, share_action))
+ recipients_by_group = _group_recipients_batch(
+ organization, groups, retained, revoked_at
+ )
+ return _mail_all_groups(
+ service, groups, recipients_by_group, shared, actor, share_action
+ )
+
+
+def _resolve_share(
+ organization: Organization, actor_id: int, resource_kind: str, resource_id: str
+) -> tuple[User, _SharedResource] | None:
+ """The actor and resolved resource, or ``None`` to skip (already logged)."""
+ actor = _get_user(organization, actor_id)
+ shared = _load_resource(organization, resource_kind, resource_id)
+ if actor is None:
+ # Actor left the org between the share and the send -- routine race,
+ # not a bug.
+ logger.info(
+ "metric=group_notification_actor_left_org_total group-notification: "
+ "skipping resource share for %s/%s (actor no longer in org)",
+ resource_kind,
+ resource_id,
+ )
+ return None
+ if shared.type is None:
+ # A registered resource kind with no notification-plugin type mapping
+ # -- a real gap worth an operator's attention, unlike the actor case.
+ logger.warning(
+ "metric=group_notification_unresolved_resource_type_total "
+ "group-notification: skipping resource share for %s/%s "
+ "(resource type not registered)",
+ resource_kind,
+ resource_id,
+ )
+ return None
+ return actor, shared
+
+
+def _mail_all_groups(
+ service: Any,
+ groups: list[OrganizationGroup],
+ recipients_by_group: dict[int, list[User]],
+ shared: _SharedResource,
+ actor: User,
+ share_action: str,
+) -> bool:
+ """Send each group's copy concurrently.
+
+ Returns ``False`` (ask for redelivery) only when every attempted group
+ genuinely failed to send. A skipped group (the plugin's tri-state
+ ``None`` -- unconfigured, disabled, bad input) never counts as a failure.
+ A *partial* failure -- some groups sent, others didn't -- is deliberately
+ not retried either: redelivery would re-mail the groups that already
+ succeeded, which is the exact duplication the worker's own retry
+ classification exists to avoid. The lost group is logged instead.
+ """
+ to_mail = [g for g in groups if recipients_by_group.get(g.pk)]
+ for group in groups:
+ logger.info(
+ "group-notification: task=notify_resource_shared_with_group "
+ "group_id=%s action=%s recipient_count=%d",
+ group.pk,
+ share_action,
+ len(recipients_by_group.get(group.pk, [])),
+ )
+ if not to_mail:
+ return True
+
+ def _send(group: OrganizationGroup) -> bool | None:
+ return _mail_group(
+ service, group, recipients_by_group[group.pk], shared, actor, share_action
+ )
+
+ workers = min(len(to_mail), _MAX_CONCURRENT_GROUP_SENDS)
+ with ThreadPoolExecutor(max_workers=workers) as pool:
+ # list() first: pool.map returns a lazy generator, and all() stopping
+ # at the first False would cancel every pending future past it --
+ # groups later in the batch would silently never be mailed at all.
+ results = list(pool.map(_send, to_mail))
+ failed = [g.pk for g, r in zip(to_mail, results, strict=True) if r is False]
+ if not failed:
+ return True
+ if len(failed) == len(to_mail):
+ return False
+ logger.error(
+ "metric=group_notification_partial_failure_total failed_group_ids=%s "
+ "of %d attempted",
+ failed,
+ len(to_mail),
+ )
+ return True
+
+
+def send_membership_changed(
+ *,
+ organization: Organization,
+ group_id: int,
+ actor_id: int,
+ membership_action: str,
+ user_ids: Iterable[int],
+) -> bool:
+ """Mail the users whose membership of ``group_id`` just changed.
+
+ Recipients are re-validated against ``OrganizationMember`` — this is where
+ the offboarding race closes, for removals as well as additions: leaving a
+ group does not remove someone from the org, so both directions validate the
+ same way.
+
+ Returns:
+ ``False`` only when there were real recipients and the plugin reported
+ a send failure. See :func:`send_resource_shared` for the full contract.
+ """
+ service = _service()
+ if service is None:
+ return True
+ actor = _get_user(organization, actor_id)
+ group = _groups_in_org(organization, [group_id]).first()
+ if actor is None or group is None:
+ logger.info(
+ "group-notification: skipping membership change for group %s "
+ "(actor_found=%s group_found=%s)",
+ group_id,
+ actor is not None,
+ group is not None,
+ )
+ return True
+ recipients = _live_member_users(organization, user_ids)
+ logger.info(
+ "group-notification: task=%s group_id=%s action=%s recipient_count=%d",
+ "notify_group_membership_changed",
+ group.pk,
+ membership_action,
+ len(recipients),
+ )
+ if not recipients:
+ return True
+ result = service.send_group_membership_notification(
+ group_name=group.name,
+ membership_action=MembershipAction(membership_action).value,
+ recipients=recipients,
+ actor=actor,
+ organization=organization,
+ )
+ # Tri-state from the plugin: None (skipped -- unconfigured, disabled, bad
+ # input) is not a failure, only an explicit False is.
+ return result is not False
+
+
+def _service() -> Any | None:
+ """The cloud email service, or ``None`` when the plugin is absent (OSS)."""
+ if not notification_plugin:
+ # An absent plugin is normal in OSS. Absent while email is switched on
+ # can only be a broken build, and the plugin loader swallows the import
+ # error at DEBUG, so this is the only place it can surface.
+ if getattr(settings, "ENABLE_EMAIL_NOTIFICATIONS", False):
+ logger.warning(
+ "group-notification: email is enabled but the notification "
+ "plugin did not load — no mail is being sent"
+ )
+ else:
+ logger.debug("group-notification: notification plugin unavailable, skipping")
+ return None
+ return notification_plugin["service_class"]()
+
+
+def _get_user(organization: Organization, user_id: int) -> User | None:
+ """The actor, re-validated against the org like every recipient is.
+
+ Service accounts are kept: a share performed by a platform account must
+ still notify the group.
+ """
+ member = (
+ OrganizationMember.objects.filter(organization=organization, user_id=user_id)
+ .select_related("user")
+ .first()
+ )
+ return member.user if member else None
+
+
+def _retained_user_ids(
+ organization: Organization, resource: Any, share_action: str
+) -> set[int] | None:
+ """Users who still reach ``resource``; empty on the share direction.
+
+ A revoked group's members may keep access by a route the revoke did not
+ touch — another group, a direct share, ownership, or being an org admin,
+ who reaches every resource in the org. Telling any of them their access was
+ removed would be wrong, and the revoke email also repoints their CTA at the
+ dashboard. ``compute_effective_members`` covers the share routes only, so
+ owners and admins are added back explicitly.
+
+ ``None`` means access never depended on the share at all, so the caller
+ skips the fan-out entirely.
+ """
+ if share_action != ShareAction.REVOKED.value:
+ return set()
+ from tenant_account_v2.sharing_helpers import retained_user_ids
+
+ retained = retained_user_ids(resource, organization)
+ if retained is None:
+ logger.info(
+ "group-notification: revoke on %s, whose access does not depend on "
+ "shares — nobody lost access, no mail",
+ resource.pk,
+ )
+ return retained
+
+
+def _groups_to_mail(
+ organization: Organization,
+ group_ids: Iterable[int],
+ resource: Any,
+ share_action: str,
+) -> Iterable[OrganizationGroup]:
+ """Groups from the payload that should still be mailed.
+
+ On a grant, drop any group whose access was revoked between enqueue and
+ delivery: the mail carries the resource name and id, so announcing access
+ the group no longer holds discloses both to members who cannot reach it.
+ The revoke direction needs no such check — its share row is already gone,
+ and ``_retained_user_ids`` covers who kept access another way.
+ """
+ groups = _groups_in_org(organization, group_ids)
+ if share_action != ShareAction.SHARED.value:
+ return groups
+ from tenant_account_v2.sharing_helpers import get_resource_share_groups
+
+ live = {group.pk for group in get_resource_share_groups(resource)}
+ to_mail = [group for group in groups if group.pk in live]
+ if len(to_mail) != len(groups):
+ logger.info(
+ "group-notification: dropped %d of %d groups (access revoked since enqueue)",
+ len(groups) - len(to_mail),
+ len(groups),
+ )
+ return to_mail
+
+
+def _group_recipients_batch(
+ organization: Organization,
+ groups: list[OrganizationGroup],
+ retained: set[int],
+ joined_before: datetime | None = None,
+) -> dict[int, list[User]]:
+ """Live members of each of ``groups`` who did not keep access via ``retained``.
+
+ Two queries total (one ``GroupMembership`` scan, one ``OrganizationMember``
+ validation) across every group in the fan-out, rather than one pair per
+ group -- a resource shared with N groups issued N pairs of queries before
+ this, since ``joined_before`` (a revoke's timestamp) is the same
+ cutoff for every group being mailed in one call, so the membership lookup
+ batches cleanly.
+
+ ``joined_before`` drops anyone who joined after the access was taken away:
+ they never held it through this group, so a revocation notice would be
+ about access they never had.
+ """
+ if not groups:
+ return {}
+ memberships = GroupMembership.objects.filter(group__in=groups)
+ if joined_before is not None:
+ memberships = memberships.filter(created_at__lte=joined_before)
+ user_ids_by_group: dict[int, set[int]] = defaultdict(set)
+ all_user_ids: set[int] = set()
+ for group_id, user_id in memberships.values_list("group_id", "user_id"):
+ user_ids_by_group[group_id].add(user_id)
+ all_user_ids.add(user_id)
+ users_by_id = {
+ user.pk: user for user in _live_member_users(organization, all_user_ids)
+ }
+ return {
+ group.pk: [
+ users_by_id[uid]
+ for uid in user_ids_by_group.get(group.pk, ())
+ if uid in users_by_id and uid not in retained
+ ]
+ for group in groups
+ }
+
+
+def _mail_group(
+ service: Any,
+ group: OrganizationGroup,
+ recipients: list[User],
+ shared: _SharedResource,
+ actor: User,
+ share_action: str,
+) -> bool | None:
+ """Send one group's copy of the resource-share email.
+
+ Passes through the plugin's tri-state result -- see
+ :func:`_mail_all_groups` for how ``None`` (skipped) is distinguished
+ from ``False`` (genuinely failed).
+ """
+ return service.send_group_resource_shared_notification(
+ resource_type=shared.type,
+ resource_name=shared.name,
+ resource_id=str(shared.instance.pk),
+ group_name=group.name,
+ shared_by=actor,
+ shared_to=recipients,
+ resource_instance=shared.instance,
+ share_action=ShareAction(share_action).value,
+ )
+
+
+def _groups_in_org(
+ organization: Organization, group_ids: Iterable[int]
+) -> QuerySet[OrganizationGroup]:
+ """Groups from ``group_ids`` that belong to ``organization``."""
+ return OrganizationGroup.objects.filter(
+ organization=organization, pk__in=list(group_ids)
+ )
+
+
+def _live_member_users(organization: Organization, user_ids: Iterable[int]) -> list[User]:
+ """Users from ``user_ids`` who are still live members of ``organization``.
+
+ Service accounts are excluded, matching ``compute_effective_members``. Also
+ drops anyone with a falsy ``email`` -- silently, since this list decides
+ who has real recipients, and that in turn decides whether a group is
+ attempted at all (and so whether a 502 can ever fire for it).
+ """
+ requested = list(user_ids)
+ memberships = OrganizationMember.objects.filter(
+ organization=organization, user_id__in=requested
+ ).select_related("user")
+ users = [
+ m.user
+ for m in memberships
+ if not getattr(m.user, "is_service_account", False) and m.user.email
+ ]
+ if len(users) != len(requested):
+ logger.info(
+ "group-notification: dropped %d of %d recipients "
+ "(left the org / service account / no email)",
+ len(requested) - len(users),
+ len(requested),
+ )
+ return users
+
+
+def _load_resource(
+ organization: Organization, kind: str, resource_id: str
+) -> _SharedResource:
+ """Resolve the shared resource for the email senders.
+
+ Raises:
+ ResourceNotFoundError: the descriptor, model, or row is missing — the
+ resource was deleted or belongs to another org. Callers turn this
+ into a success so the queue stops retrying.
+ """
+ descriptor = descriptor_for_kind(kind)
+ if descriptor is None:
+ raise ResourceNotFoundError(f"Unknown resource kind: {kind}")
+ try:
+ model = apps.get_model(descriptor.app_label, descriptor.model_name)
+ except LookupError as exc: # cloud-only app not installed here
+ raise ResourceNotFoundError(f"Model unavailable for kind: {kind}") from exc
+ # Filter on the organization explicitly rather than trusting the default
+ # manager: ``AgenticProject``'s manager deliberately spans organizations.
+ resource = model.objects.filter(
+ organization=organization, **{descriptor.id_field: resource_id}
+ ).first()
+ if resource is None:
+ raise ResourceNotFoundError(f"{kind} {resource_id} not found in organization")
+ # Populate the FK cache with the instance we already hold: the mail send
+ # (``resource_instance.organization``) runs inside a pool thread, and a
+ # lazy query there opens a connection ``close_old_connections`` never
+ # cleans up (that hook only runs on the request thread).
+ resource.organization = organization
+ name = getattr(resource, descriptor.name_field, "") or ""
+ return _SharedResource(resource, name, _resource_type_for(descriptor, resource))
+
+
+def _resource_type_for(descriptor: ShareableResource, resource: Any) -> str | None:
+ """Map a resource to the email plugin's ``ResourceType`` value.
+
+ Returns ``None`` for resources the plugin has no type for (e.g. a pipeline
+ that is neither ETL nor TASK) — the caller skips rather than guessing.
+ """
+ if descriptor.kind == "pipeline":
+ return pipeline_notification_type(getattr(resource, "pipeline_type", None))
+ if descriptor.kind == "adapter_instance":
+ return adapter_notification_type(str(resource.adapter_type or ""))
+ return _STATIC_RESOURCE_TYPES.get(descriptor.kind)
diff --git a/backend/tenant_account_v2/group_views.py b/backend/tenant_account_v2/group_views.py
index 76461951fa..4814f883f7 100644
--- a/backend/tenant_account_v2/group_views.py
+++ b/backend/tenant_account_v2/group_views.py
@@ -26,6 +26,10 @@
GroupMembership,
OrganizationGroup,
)
+from tenant_account_v2.share_notifications import (
+ MembershipAction,
+ notify_group_membership_changed,
+)
logger = logging.getLogger(__name__)
@@ -151,12 +155,30 @@ def members(self, request: Request, pk: str | None = None) -> Response:
serializer = GroupMemberAddSerializer(data=request.data, context={"group": group})
serializer.is_valid(raise_exception=True)
user_ids_to_add: list[int] = serializer.validated_data["user_ids_to_add"]
+ # The serializer's "already a member" check ran at validation time, not
+ # at this write, so a concurrent request adding the same user in
+ # between would make our own insert a silent no-op (ignore_conflicts)
+ # while we still believe we added them. Re-check right before the
+ # write to keep that window as small as it can be.
+ already_members = set(
+ group.memberships.filter(user_id__in=user_ids_to_add).values_list(
+ "user_id", flat=True
+ )
+ )
+ newly_added_ids = [uid for uid in user_ids_to_add if uid not in already_members]
GroupMembership.objects.bulk_create(
- [GroupMembership(group=group, user_id=uid) for uid in user_ids_to_add],
+ [GroupMembership(group=group, user_id=uid) for uid in newly_added_ids],
ignore_conflicts=True,
)
+ if newly_added_ids:
+ notify_group_membership_changed(
+ group=group,
+ action=MembershipAction.ADDED,
+ user_ids=newly_added_ids,
+ actor=request.user,
+ )
return Response(
- {"added_user_ids": user_ids_to_add},
+ {"added_user_ids": newly_added_ids},
status=status.HTTP_201_CREATED,
)
@@ -178,6 +200,12 @@ def remove_member(
deleted, _ = group.memberships.filter(user_id=user_id_int).delete()
if not deleted:
raise NotFound("User is not a member of this group.")
+ notify_group_membership_changed(
+ group=group,
+ action=MembershipAction.REMOVED,
+ user_ids=[user_id_int],
+ actor=request.user,
+ )
return Response(status=status.HTTP_204_NO_CONTENT)
# --- resources shared with this group ------------------------------------
diff --git a/backend/tenant_account_v2/internal_urls.py b/backend/tenant_account_v2/internal_urls.py
new file mode 100644
index 0000000000..4049c761a5
--- /dev/null
+++ b/backend/tenant_account_v2/internal_urls.py
@@ -0,0 +1,20 @@
+"""Internal API URLs for group-sharing email notifications."""
+
+from django.urls import path
+
+from . import internal_views
+
+app_name = "group_notification_internal"
+
+urlpatterns = [
+ path(
+ "resource-shared/",
+ internal_views.ResourceSharedWithGroupView.as_view(),
+ name="resource-shared",
+ ),
+ path(
+ "membership-changed/",
+ internal_views.GroupMembershipChangedView.as_view(),
+ name="membership-changed",
+ ),
+]
diff --git a/backend/tenant_account_v2/internal_views.py b/backend/tenant_account_v2/internal_views.py
new file mode 100644
index 0000000000..434ee453e1
--- /dev/null
+++ b/backend/tenant_account_v2/internal_views.py
@@ -0,0 +1,103 @@
+"""Internal API views for group-sharing email notifications.
+
+Mounted under ``/internal/`` and gated by ``InternalAPIAuthMiddleware``. The
+notification worker calls these because ``workers/`` has no Django and every
+step of the send — group expansion, org re-validation, resource lookup, the
+email plugin — needs it.
+
+A 200 does not imply the email was sent -- see the send path in
+:mod:`tenant_account_v2.group_notification_service`.
+"""
+
+import logging
+
+from account_v2.models import Organization
+from rest_framework import serializers, status
+from rest_framework.exceptions import ValidationError
+from rest_framework.request import Request
+from rest_framework.response import Response
+from rest_framework.views import APIView
+from utils.user_context import UserContext
+
+from tenant_account_v2.group_notification_service import (
+ ResourceNotFoundError,
+ send_membership_changed,
+ send_resource_shared,
+)
+from tenant_account_v2.share_notifications import MembershipAction, ShareAction
+
+logger = logging.getLogger(__name__)
+
+
+class ResourceSharedWithGroupSerializer(serializers.Serializer):
+ """Payload of ``notify_resource_shared_with_group``."""
+
+ group_ids = serializers.ListField(child=serializers.IntegerField(), allow_empty=False)
+ actor_id = serializers.IntegerField()
+ resource_kind = serializers.CharField()
+ resource_id = serializers.CharField()
+ # Required, and the worker task takes no default for it either: a default
+ # on both sides would turn a dropped field into a revoke mailed as a share.
+ share_action = serializers.ChoiceField(choices=[a.value for a in ShareAction])
+ # Revoke only: members who joined after this are excluded from the mail.
+ # Nullable because the worker sends the key on both directions.
+ revoked_at = serializers.DateTimeField(allow_null=True)
+
+
+class GroupMembershipChangedSerializer(serializers.Serializer):
+ """Payload of ``notify_group_membership_changed``."""
+
+ group_id = serializers.IntegerField()
+ actor_id = serializers.IntegerField()
+ membership_action = serializers.ChoiceField(
+ choices=[a.value for a in MembershipAction]
+ )
+ user_ids = serializers.ListField(child=serializers.IntegerField(), allow_empty=False)
+
+
+class _GroupNotificationView(APIView):
+ """Shared org resolution for the group-notification endpoints."""
+
+ @staticmethod
+ def _organization() -> Organization:
+ organization = UserContext.get_organization()
+ if organization is None:
+ raise ValidationError(
+ "Organization context missing. Worker must send X-Organization-ID."
+ )
+ return organization
+
+
+class ResourceSharedWithGroupView(_GroupNotificationView):
+ """Mail every current member of the groups whose resource access just changed."""
+
+ def post(self, request: Request) -> Response:
+ serializer = ResourceSharedWithGroupSerializer(data=request.data)
+ serializer.is_valid(raise_exception=True)
+ data = serializer.validated_data
+ try:
+ sent = send_resource_shared(organization=self._organization(), **data)
+ except ResourceNotFoundError as exc:
+ # Deleted between the share and the send — a retry cannot help.
+ logger.info("group-notification: dropping resource share (%s)", exc)
+ return Response({"status": "skipped"}, status=status.HTTP_200_OK)
+ if not sent:
+ # A group had real recipients and the plugin failed to send --
+ # non-2xx so the worker's retry loop (and PG redelivery once that's
+ # exhausted) actually fires, instead of a lost email reporting green.
+ return Response({"status": "failed"}, status=status.HTTP_502_BAD_GATEWAY)
+ return Response({"status": "success"}, status=status.HTTP_200_OK)
+
+
+class GroupMembershipChangedView(_GroupNotificationView):
+ """Mail the users whose group membership just changed."""
+
+ def post(self, request: Request) -> Response:
+ serializer = GroupMembershipChangedSerializer(data=request.data)
+ serializer.is_valid(raise_exception=True)
+ sent = send_membership_changed(
+ organization=self._organization(), **serializer.validated_data
+ )
+ if not sent:
+ return Response({"status": "failed"}, status=status.HTTP_502_BAD_GATEWAY)
+ return Response({"status": "success"}, status=status.HTTP_200_OK)
diff --git a/backend/tenant_account_v2/notification_resource_types.py b/backend/tenant_account_v2/notification_resource_types.py
new file mode 100644
index 0000000000..ef4a32ae49
--- /dev/null
+++ b/backend/tenant_account_v2/notification_resource_types.py
@@ -0,0 +1,34 @@
+"""Resource-type mapping shared between the direct-share and group-share
+notification paths, so a new adapter or pipeline type only needs registering
+once.
+"""
+
+from __future__ import annotations
+
+
+def adapter_notification_type(adapter_type: str) -> str:
+ """Map an ``AdapterInstance.adapter_type`` to the plugin's ``ResourceType``.
+
+ Unknown types fall back to ``LLM`` so a newly added adapter kind still
+ mails rather than silently going quiet.
+ """
+ from plugins.notification.constants import ResourceType
+
+ return {
+ "LLM": ResourceType.LLM.value,
+ "EMBEDDING": ResourceType.EMBEDDING.value,
+ "VECTOR_DB": ResourceType.VECTOR_DB.value,
+ "X2TEXT": ResourceType.X2TEXT.value,
+ }.get(adapter_type, ResourceType.LLM.value)
+
+
+def pipeline_notification_type(pipeline_type: str | None) -> str | None:
+ """Map a ``Pipeline.pipeline_type`` to the plugin's ``ResourceType``.
+
+ Only ETL/TASK pipelines are notifiable; anything else returns ``None``.
+ """
+ from plugins.notification.constants import ResourceType
+
+ if pipeline_type in (ResourceType.ETL.value, ResourceType.TASK.value):
+ return pipeline_type
+ return None
diff --git a/backend/tenant_account_v2/share_notifications.py b/backend/tenant_account_v2/share_notifications.py
new file mode 100644
index 0000000000..6b68301670
--- /dev/null
+++ b/backend/tenant_account_v2/share_notifications.py
@@ -0,0 +1,249 @@
+"""Enqueue hooks for group-sharing email notifications.
+
+Two events earn a group's members an email: a resource shared with or revoked
+from the group, and a user added to or removed from it. Both are dispatched
+asynchronously — the caller's request returns as soon as the write lands.
+
+The sending itself runs in ``workers/``, which is Django-free, so the worker
+task is a thin HTTP shim back to :mod:`tenant_account_v2.internal_views`; the
+backend does the ORM and plugin work. Transport is the PG queue, the only
+one there is — the ``notifications`` queue the notification consumer polls.
+
+A missing org, an absent notification plugin, or any dispatch error means no
+notification, never a broken share. Sending is also bounded by the cloud
+``ENABLE_EMAIL_NOTIFICATIONS`` setting and by each event's own template ID
+being configured -- either being unset silently skips, same as the
+direct-user mail this reuses.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Iterable
+from enum import StrEnum
+from typing import TYPE_CHECKING, Any
+
+from django.utils import timezone
+from plugins import get_plugin
+
+from tenant_account_v2.shareable_resources import kind_for_instance
+
+if TYPE_CHECKING:
+ from account_v2.models import User
+
+ from tenant_account_v2.models import OrganizationGroup
+
+logger = logging.getLogger(__name__)
+
+NOTIFY_RESOURCE_SHARED_TASK = "notify_resource_shared_with_group"
+NOTIFY_MEMBERSHIP_CHANGED_TASK = "notify_group_membership_changed"
+
+# Mirrors the workers' ``QueueName.NOTIFICATION`` — a local literal so the
+# backend does not import the workers package (same as ``pipeline_dispatch``).
+NOTIFICATION_QUEUE = "notifications"
+
+
+class MembershipAction(StrEnum):
+ """What happened to a user's membership of a group."""
+
+ ADDED = "added"
+ REMOVED = "removed"
+
+
+class ShareAction(StrEnum):
+ """What happened to a group's access to a resource."""
+
+ SHARED = "shared"
+ REVOKED = "revoked"
+
+
+def notify_resource_group_share_changed(
+ *,
+ resource: Any,
+ added: Iterable[OrganizationGroup],
+ removed: Iterable[OrganizationGroup],
+ actor: User,
+) -> None:
+ """Queue group mail for a resource just shared with / revoked from groups."""
+ for share_action, groups in (
+ (ShareAction.SHARED, added),
+ (ShareAction.REVOKED, removed),
+ ):
+ _notify_group_share(
+ resource=resource, groups=groups, share_action=share_action, actor=actor
+ )
+
+
+def _notify_group_share(
+ *,
+ resource: Any,
+ groups: Iterable[OrganizationGroup],
+ share_action: ShareAction,
+ actor: User,
+) -> None:
+ """Queue one group-share event.
+
+ Recipients are resolved at delivery time rather than frozen here: anyone
+ who leaves the org between the click and the send simply isn't in the fresh
+ lookup, so offboarding safety costs nothing. Unlike a membership removal,
+ revoking a group's access leaves the group and its members intact, so the
+ fresh lookup still finds everyone who needs telling.
+
+ A revoke carries ``revoked_at`` so that fresh lookup can still exclude
+ anyone who joined the group *after* the access was taken away — they never
+ held it through this group, and the queue can lag. One timestamp rather
+ than the whole member list, which would grow the payload with the group.
+ """
+ group_ids = sorted(group.pk for group in groups)
+ if not group_ids:
+ return
+ # Stamped at the moment of the revoke, not at delivery: a member joining
+ # after it would otherwise be mailed about access they never held.
+ revoked_at = (
+ timezone.now().isoformat() if share_action is ShareAction.REVOKED else None
+ )
+ organization_id = _organization_slug(resource)
+ kind = kind_for_instance(resource)
+ if not organization_id or kind is None:
+ # Neither is a routine skip: a shareable resource always resolves an
+ # org, and the share endpoint accepts ``shared_groups`` for any host
+ # viewset, so an unregistered kind means a group share landed on a
+ # resource this feature cannot mail about.
+ logger.warning(
+ "group-notification: skipping %s share for %s %s (organization=%s kind=%s)",
+ share_action,
+ type(resource).__name__,
+ resource.pk,
+ organization_id,
+ kind,
+ )
+ return
+ kwargs: dict[str, Any] = {
+ "group_ids": group_ids,
+ "actor_id": actor.pk,
+ "resource_kind": kind,
+ "resource_id": str(resource.pk),
+ "share_action": share_action.value,
+ "organization_id": organization_id,
+ }
+ if revoked_at is not None:
+ kwargs["revoked_at"] = revoked_at
+ _dispatch_quietly(
+ task_name=NOTIFY_RESOURCE_SHARED_TASK,
+ kwargs=kwargs,
+ organization_id=organization_id,
+ )
+
+
+def notify_group_membership_changed(
+ *,
+ group: OrganizationGroup,
+ action: MembershipAction,
+ user_ids: Iterable[int],
+ actor: User,
+) -> None:
+ """Queue "you were added to / removed from a group" mail for those users.
+
+ Unlike a resource share, the user ids ride in the payload: on removal the
+ membership rows are already gone by delivery time, and on add a fresh group
+ lookup would mail every existing member too.
+ """
+ recipients = sorted(user_ids)
+ if not recipients:
+ return
+ organization_id = _organization_slug(group)
+ if not organization_id:
+ logger.warning(
+ "group-notification: skipping membership change for group %s "
+ "(no resolvable organization)",
+ group.pk,
+ )
+ return
+ _dispatch_quietly(
+ task_name=NOTIFY_MEMBERSHIP_CHANGED_TASK,
+ kwargs={
+ "group_id": group.pk,
+ "actor_id": actor.pk,
+ "membership_action": action.value,
+ "user_ids": recipients,
+ "organization_id": organization_id,
+ },
+ organization_id=organization_id,
+ )
+
+
+def _organization_slug(obj: Any) -> str | None:
+ """The owning org's string identifier (``Organization.organization_id``).
+
+ This is the ``X-Organization-ID`` value the worker echoes back, not the DB
+ pk, and it is what the queue row records for fairness/routing.
+ """
+ organization = getattr(obj, "organization", None)
+ return getattr(organization, "organization_id", None)
+
+
+def _dispatch_quietly(
+ *,
+ task_name: str,
+ kwargs: dict[str, Any],
+ organization_id: str,
+) -> None:
+ """Enqueue on the PG queue; never let a failure reach the caller.
+
+ The share or membership change has already been committed by the time this
+ runs — losing its email is not a reason to fail the request the user made.
+
+ Skips the enqueue entirely when the notification plugin isn't loaded (pure
+ OSS): the worker would just claim the row and no-op, so there is no point
+ writing a queue row nobody can act on.
+ """
+ if not get_plugin("notification"):
+ logger.debug(
+ "group-notification: notification plugin unavailable, skipping "
+ "enqueue for %s",
+ task_name,
+ )
+ return
+ try:
+ _dispatch(
+ task_name=task_name,
+ kwargs=kwargs,
+ organization_id=organization_id,
+ )
+ except Exception:
+ logger.exception(
+ "metric=group_notification_enqueue_failed_total task=%s org_id=%s",
+ task_name,
+ organization_id,
+ )
+
+
+def _dispatch(
+ *,
+ task_name: str,
+ kwargs: dict[str, Any],
+ organization_id: str,
+) -> None:
+ """Enqueue on the PG queue.
+
+ Deploy the notification worker at or before the backend: the consumer polls
+ the queue by name, so a pod still on the previous image claims these rows,
+ cannot resolve the task, and DELETES them (no dead-letter — nothing here
+ passes ``reply_key`` or ``on_error``).
+ """
+ # Lazy import — ``pg_queue`` is heavier than this leaf module and importing
+ # it at load time risks a cycle during Django app loading.
+ from pg_queue.producer import enqueue_task
+
+ msg_id = enqueue_task(
+ task_name=task_name,
+ queue=NOTIFICATION_QUEUE,
+ kwargs=kwargs,
+ org_id=organization_id,
+ )
+ logger.info(
+ "group-notification: %s enqueued on PG queue %r (msg_id=%s)",
+ task_name,
+ NOTIFICATION_QUEUE,
+ msg_id,
+ )
diff --git a/backend/tenant_account_v2/shareable_resources.py b/backend/tenant_account_v2/shareable_resources.py
index 80e7356084..ba8b02573c 100644
--- a/backend/tenant_account_v2/shareable_resources.py
+++ b/backend/tenant_account_v2/shareable_resources.py
@@ -9,6 +9,7 @@
"""
from dataclasses import dataclass
+from typing import Any
@dataclass(frozen=True)
@@ -50,3 +51,26 @@ class ShareableResource:
),
ShareableResource("lookups", "LookupDefinition", "lookup", "name", "lookup_id"),
)
+
+
+def descriptor_for_kind(kind: str) -> ShareableResource | None:
+ """Look up a descriptor by its ``kind`` key."""
+ return next((r for r in SHAREABLE_RESOURCES if r.kind == kind), None)
+
+
+def kind_for_instance(instance: Any) -> str | None:
+ """Reverse lookup: the ``kind`` of a resource instance, ``None`` if unlisted.
+
+ Matches on the model's app label + class name so callers holding an
+ instance (e.g. the share endpoint) don't hardcode a type check per
+ resource.
+ """
+ meta = instance._meta
+ return next(
+ (
+ r.kind
+ for r in SHAREABLE_RESOURCES
+ if r.app_label == meta.app_label and r.model_name == meta.object_name
+ ),
+ None,
+ )
diff --git a/backend/tenant_account_v2/sharing_helpers.py b/backend/tenant_account_v2/sharing_helpers.py
index e7035e59d8..c7775bb682 100644
--- a/backend/tenant_account_v2/sharing_helpers.py
+++ b/backend/tenant_account_v2/sharing_helpers.py
@@ -269,6 +269,38 @@ def serialize_owner_refs(resource_obj: Any) -> list[dict[str, Any]]:
return [{"id": user.pk, "email": user.email} for user in resource_obj.owners()]
+def access_survives_share_changes(resource_obj: Any) -> bool:
+ """True when access to ``resource_obj`` does not depend on shares at all.
+
+ Revoking a share removes nothing in that case, so nobody should be told it
+ did. ``shared_to_org`` covers every org member; ``is_friction_less`` is the
+ adapter equivalent -- ``AdapterInstance.for_user`` admits it to any regular
+ user unconditionally (service accounts are the one exception, explicitly
+ excluded from frictionless adapters).
+ """
+ return bool(getattr(resource_obj, "shared_to_org", False)) or bool(
+ getattr(resource_obj, "is_friction_less", False)
+ )
+
+
+def org_admin_user_ids(organization: Any) -> set[int]:
+ """Users who reach every resource in ``organization`` as admins.
+
+ ``for_user`` returns the whole queryset for an org admin, so a revoke never
+ takes their access away and they must not be mailed about losing it. The
+ role STRING is plugin-dependent (it differs between the OSS and auth0
+ plugins), so this goes through the auth controller rather than comparing to
+ a literal.
+ """
+ from account_v2.authentication_controller import AuthenticationController
+
+ controller = AuthenticationController()
+ rows = OrganizationMember.objects.filter(organization=organization).values_list(
+ "user_id", "role"
+ )
+ return {uid for uid, role in rows if controller.is_admin_by_role(role)}
+
+
def compute_effective_members(resource_obj: Any) -> list[dict[str, Any]]:
"""Compute effective members of a shareable resource.
@@ -321,6 +353,30 @@ def compute_effective_members(resource_obj: Any) -> list[dict[str, Any]]:
return list(seen.values())
+def retained_user_ids(resource_obj: Any, organization: Any = None) -> set[int] | None:
+ """Every user id who still reaches ``resource_obj`` after a revoke.
+
+ ``None`` means access never depended on shares at all (see
+ ``access_survives_share_changes``) -- the caller's cue that nobody lost
+ anything and no notification is owed. Shared by the direct-share and
+ group-share revoke paths so they can't independently drift on what counts
+ as "still has access".
+ """
+ if access_survives_share_changes(resource_obj):
+ return None
+ retained = {member["user_id"] for member in compute_effective_members(resource_obj)}
+ # compute_effective_members deliberately excludes owners (they hold the
+ # resource, they aren't "shared with" it); a revoke never touches them.
+ retained |= {owner.pk for owner in resource_obj.owners()}
+ org = (
+ organization
+ if organization is not None
+ else getattr(resource_obj, "organization", None)
+ )
+ retained |= org_admin_user_ids(org)
+ return retained
+
+
def _add_org_members(seen: dict[int, dict[str, Any]], resource_obj: Any) -> None:
"""Add org-wide members to ``seen`` (skips users already recorded)."""
if not getattr(resource_obj, "shared_to_org", False):
diff --git a/backend/tenant_account_v2/test_internal_views.py b/backend/tenant_account_v2/test_internal_views.py
new file mode 100644
index 0000000000..b936bf097b
--- /dev/null
+++ b/backend/tenant_account_v2/test_internal_views.py
@@ -0,0 +1,93 @@
+"""View-level tests for the group-notification internal endpoints.
+
+The send-side logic (who gets mailed, the tri-state plugin contract) is
+covered in ``ResourceShareNotificationTests`` in ``tests.py``. These pin the
+one thing that lives only here: the status-code mapping from that result to
+an HTTP response, since that mapping is what the worker's retry decision
+actually reads. ``send_resource_shared`` / ``send_membership_changed`` are
+patched directly so these don't need a full group/resource DB setup to
+exercise the view in isolation.
+"""
+
+from unittest.mock import patch
+
+from account_v2.models import Organization
+from django.test import RequestFactory, TestCase
+from rest_framework import status
+from utils.user_context import UserContext
+
+from tenant_account_v2.internal_views import (
+ GroupMembershipChangedView,
+ ResourceSharedWithGroupView,
+)
+
+_SHARE_PAYLOAD = {
+ "group_ids": [1],
+ "actor_id": 1,
+ "resource_kind": "workflow",
+ "resource_id": "wf-1",
+ "share_action": "shared",
+ "revoked_at": None,
+}
+
+_MEMBERSHIP_PAYLOAD = {
+ "group_id": 1,
+ "actor_id": 1,
+ "membership_action": "added",
+ "user_ids": [1],
+}
+
+
+class _InternalViewTestBase(TestCase):
+ def setUp(self) -> None:
+ self.org = Organization.objects.create(
+ name="org-views", display_name="Org Views", organization_id="org-views"
+ )
+ UserContext.set_organization_identifier(self.org.organization_id)
+ self.addCleanup(UserContext.set_organization_identifier, None)
+
+ @staticmethod
+ def _post(view_cls, data: dict):
+ request = RequestFactory().post(
+ "/internal/", data=data, content_type="application/json"
+ )
+ return view_cls.as_view()(request)
+
+
+class ResourceSharedWithGroupViewTests(_InternalViewTestBase):
+ def test_sent_returns_200(self) -> None:
+ with patch(
+ "tenant_account_v2.internal_views.send_resource_shared", return_value=True
+ ):
+ response = self._post(ResourceSharedWithGroupView, _SHARE_PAYLOAD)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(response.data["status"], "success")
+
+ def test_genuine_failure_returns_502(self) -> None:
+ # This is the branch the Critical hinged on: before the tri-state
+ # fix, every skip (unconfigured template, disabled notifications,
+ # bad input) collapsed into this same False -- retrying forever a
+ # condition no retry could fix.
+ with patch(
+ "tenant_account_v2.internal_views.send_resource_shared", return_value=False
+ ):
+ response = self._post(ResourceSharedWithGroupView, _SHARE_PAYLOAD)
+ self.assertEqual(response.status_code, status.HTTP_502_BAD_GATEWAY)
+ self.assertEqual(response.data["status"], "failed")
+
+
+class GroupMembershipChangedViewTests(_InternalViewTestBase):
+ def test_sent_returns_200(self) -> None:
+ with patch(
+ "tenant_account_v2.internal_views.send_membership_changed", return_value=True
+ ):
+ response = self._post(GroupMembershipChangedView, _MEMBERSHIP_PAYLOAD)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+ def test_genuine_failure_returns_502(self) -> None:
+ with patch(
+ "tenant_account_v2.internal_views.send_membership_changed",
+ return_value=False,
+ ):
+ response = self._post(GroupMembershipChangedView, _MEMBERSHIP_PAYLOAD)
+ self.assertEqual(response.status_code, status.HTTP_502_BAD_GATEWAY)
diff --git a/backend/tenant_account_v2/test_share_notification_dispatch.py b/backend/tenant_account_v2/test_share_notification_dispatch.py
new file mode 100644
index 0000000000..f711a3457d
--- /dev/null
+++ b/backend/tenant_account_v2/test_share_notification_dispatch.py
@@ -0,0 +1,161 @@
+"""Unit tests for the group-notification enqueue side (UN-3494 / mfbt UNS-848).
+
+``share_notifications`` runs inside the user's share request: it builds the task
+payload and hands it to the transport. Nothing here touches the ORM or sends
+mail, so the module is patched at its three seams — ``kind_for_instance``,
+``get_plugin`` and ``_dispatch`` — and these run in the rig's unit tier with no
+Postgres. The transport itself is covered by ``pg_queue.tests``; the
+delivery side by ``ResourceShareNotificationTests`` in ``tenant_account_v2.tests``.
+"""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import tenant_account_v2.share_notifications as sn
+
+_ACTOR = SimpleNamespace(pk=7)
+_RESOURCE = SimpleNamespace(
+ pk="wf-1", organization=SimpleNamespace(organization_id="org-a")
+)
+
+
+def _group(pk: int) -> SimpleNamespace:
+ return SimpleNamespace(pk=pk)
+
+
+@contextmanager
+def _seams(*, dispatch_raises: Exception | None = None):
+ """Patch the module's three outbound seams; yield the dispatch mock."""
+ with (
+ patch.object(sn, "kind_for_instance", return_value="workflow"),
+ patch.object(sn, "get_plugin", return_value={"service_class": object()}),
+ patch.object(sn, "_dispatch", side_effect=dispatch_raises) as dispatch,
+ ):
+ yield dispatch
+
+
+class TestNotifyResourceGroupShareChanged:
+ def test_grant_dispatches_shared_payload_without_timestamp(self):
+ with _seams() as dispatch:
+ sn.notify_resource_group_share_changed(
+ resource=_RESOURCE, added=[_group(5), _group(2)], removed=[], actor=_ACTOR
+ )
+ dispatch.assert_called_once()
+ call = dispatch.call_args.kwargs
+ assert call["task_name"] == sn.NOTIFY_RESOURCE_SHARED_TASK
+ assert call["organization_id"] == "org-a"
+ assert call["kwargs"] == {
+ "group_ids": [2, 5], # sorted, so the payload is stable
+ "actor_id": 7,
+ "resource_kind": "workflow",
+ "resource_id": "wf-1",
+ "share_action": "shared",
+ "organization_id": "org-a",
+ }
+ # A grant carries no cutoff — the delivery side mails every live member.
+ assert "revoked_at" not in call["kwargs"]
+
+ def test_revoke_dispatches_revoked_payload_with_timestamp(self):
+ with _seams() as dispatch:
+ sn.notify_resource_group_share_changed(
+ resource=_RESOURCE, added=[], removed=[_group(3)], actor=_ACTOR
+ )
+ payload = dispatch.call_args.kwargs["kwargs"]
+ assert payload["share_action"] == "revoked"
+ assert payload["group_ids"] == [3]
+ # ISO-8601 string, not a datetime — the payload is JSON-serialized.
+ datetime.fromisoformat(payload["revoked_at"])
+
+ def test_grant_and_revoke_dispatch_independently(self):
+ with _seams() as dispatch:
+ sn.notify_resource_group_share_changed(
+ resource=_RESOURCE, added=[_group(1)], removed=[_group(2)], actor=_ACTOR
+ )
+ assert dispatch.call_count == 2
+ actions = [c.kwargs["kwargs"]["share_action"] for c in dispatch.call_args_list]
+ assert actions == ["shared", "revoked"]
+
+ def test_no_groups_skips_dispatch(self):
+ with _seams() as dispatch:
+ sn.notify_resource_group_share_changed(
+ resource=_RESOURCE, added=[], removed=[], actor=_ACTOR
+ )
+ dispatch.assert_not_called()
+
+ def test_unknown_resource_kind_skips_dispatch(self):
+ with (
+ patch.object(sn, "kind_for_instance", return_value=None),
+ patch.object(sn, "get_plugin", return_value={"service_class": object()}),
+ patch.object(sn, "_dispatch") as dispatch,
+ ):
+ sn.notify_resource_group_share_changed(
+ resource=_RESOURCE, added=[_group(1)], removed=[], actor=_ACTOR
+ )
+ dispatch.assert_not_called()
+
+ def test_missing_organization_skips_dispatch(self):
+ orphan = SimpleNamespace(pk="wf-1", organization=None)
+ with _seams() as dispatch:
+ sn.notify_resource_group_share_changed(
+ resource=orphan, added=[_group(1)], removed=[], actor=_ACTOR
+ )
+ dispatch.assert_not_called()
+
+ def test_plugin_absent_skips_dispatch(self):
+ # Pure OSS: nothing would ever consume this row, so don't write it.
+ with (
+ patch.object(sn, "kind_for_instance", return_value="workflow"),
+ patch.object(sn, "get_plugin", return_value=None),
+ patch.object(sn, "_dispatch") as dispatch,
+ ):
+ sn.notify_resource_group_share_changed(
+ resource=_RESOURCE, added=[_group(1)], removed=[], actor=_ACTOR
+ )
+ dispatch.assert_not_called()
+
+ def test_dispatch_failure_never_reaches_the_caller(self):
+ # The share has already committed — losing its email must not 500 it.
+ with _seams(dispatch_raises=RuntimeError("queue down")):
+ sn.notify_resource_group_share_changed(
+ resource=_RESOURCE, added=[_group(1)], removed=[], actor=_ACTOR
+ )
+
+
+class TestNotifyGroupMembershipChanged:
+ def test_membership_change_dispatches_user_ids_in_payload(self):
+ with _seams() as dispatch:
+ sn.notify_group_membership_changed(
+ group=SimpleNamespace(
+ pk=9, organization=SimpleNamespace(organization_id="org-a")
+ ),
+ action=sn.MembershipAction.ADDED,
+ user_ids=[4, 1],
+ actor=_ACTOR,
+ )
+ call = dispatch.call_args.kwargs
+ assert call["task_name"] == sn.NOTIFY_MEMBERSHIP_CHANGED_TASK
+ assert call["kwargs"] == {
+ "group_id": 9,
+ "actor_id": 7,
+ "membership_action": "added",
+ # Unlike a share, the ids ride in the payload: on removal the rows
+ # are gone by delivery time.
+ "user_ids": [1, 4],
+ "organization_id": "org-a",
+ }
+
+ def test_no_users_skips_dispatch(self):
+ with _seams() as dispatch:
+ sn.notify_group_membership_changed(
+ group=SimpleNamespace(
+ pk=9, organization=SimpleNamespace(organization_id="org-a")
+ ),
+ action=sn.MembershipAction.REMOVED,
+ user_ids=[],
+ actor=_ACTOR,
+ )
+ dispatch.assert_not_called()
diff --git a/backend/tenant_account_v2/test_sharing_retention.py b/backend/tenant_account_v2/test_sharing_retention.py
new file mode 100644
index 0000000000..1c051eeb9c
--- /dev/null
+++ b/backend/tenant_account_v2/test_sharing_retention.py
@@ -0,0 +1,52 @@
+"""Which resources have access that does not depend on shares at all.
+
+Revoking a share from such a resource removes nothing, so nobody may be told it
+did. Two routes qualify and they are easy to miss because neither is a share
+row: ``shared_to_org`` admits every org member, and ``is_friction_less`` is the
+adapter equivalent -- ``AdapterInstance.for_user`` admits it unconditionally,
+alongside the share clauses.
+
+Pure predicate, so this runs in the rig's unit tier with no database. The
+admin route is the third one and is user-level rather than resource-level, so
+it lives in ``org_admin_user_ids`` and is covered by the DB tests.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from tenant_account_v2.sharing_helpers import access_survives_share_changes
+
+
+class TestAccessSurvivesShareChanges:
+ @pytest.mark.parametrize(
+ "resource, expected, why",
+ [
+ (SimpleNamespace(shared_to_org=True), True, "org-wide share"),
+ (SimpleNamespace(is_friction_less=True), True, "frictionless adapter"),
+ (
+ SimpleNamespace(shared_to_org=False, is_friction_less=True),
+ True,
+ "frictionless without an org share still admits everyone",
+ ),
+ (
+ SimpleNamespace(shared_to_org=True, is_friction_less=False),
+ True,
+ "org share without frictionless still admits everyone",
+ ),
+ (
+ SimpleNamespace(shared_to_org=False, is_friction_less=False),
+ False,
+ "neither route: access really does depend on the share",
+ ),
+ (
+ SimpleNamespace(),
+ False,
+ "a model carrying neither field, e.g. Workflow, is share-dependent",
+ ),
+ ],
+ )
+ def test_routes(self, resource, expected, why):
+ assert access_survives_share_changes(resource) is expected, why
diff --git a/backend/tenant_account_v2/tests.py b/backend/tenant_account_v2/tests.py
index a105593599..19f3d0ca6a 100644
--- a/backend/tenant_account_v2/tests.py
+++ b/backend/tenant_account_v2/tests.py
@@ -11,18 +11,25 @@
"""
import secrets
-from unittest.mock import patch
+from datetime import timedelta
+from unittest.mock import Mock, patch
from account_v2.models import Organization, User
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import FieldDoesNotExist
from django.test import TestCase
+from django.utils import timezone
from permissions.roles import ResourceRole
from rest_framework.exceptions import PermissionDenied
from rest_framework.test import APIRequestFactory, force_authenticate
from utils.user_context import UserContext
from workflow_manager.workflow_v2.models.workflow import Workflow
+from tenant_account_v2.group_notification_service import (
+ ResourceNotFoundError,
+ send_membership_changed,
+ send_resource_shared,
+)
from tenant_account_v2.group_views import OrganizationGroupViewSet
from tenant_account_v2.models import (
GroupMembership,
@@ -30,6 +37,7 @@
OrganizationMember,
ResourceGroupShare,
)
+from tenant_account_v2.share_notifications import MembershipAction, ShareAction
from tenant_account_v2.shareable_resources import SHAREABLE_RESOURCES
from tenant_account_v2.sharing_helpers import (
ShareAuthorizationService,
@@ -371,6 +379,32 @@ def test_service_account_can_add_members(self) -> None:
GroupMembership.objects.filter(group=self.group, user=self.outsider).exists()
)
+ def test_add_members_response_only_lists_newly_added(self) -> None:
+ # self.member is already in self.group (see GroupSharingTestBase);
+ # only self.outsider is new. The response, and the notification it
+ # feeds, must both narrow to the actual insert, not the request.
+ with patch(
+ "tenant_account_v2.group_views.notify_group_membership_changed"
+ ) as notify:
+ response = self._call(
+ {"post": "members"},
+ "post",
+ self.svc,
+ data={"user_ids": [self.member.id, self.outsider.id]},
+ pk=str(self.group.pk),
+ )
+ self.assertEqual(response.status_code, 201)
+ self.assertEqual(response.data["added_user_ids"], [self.outsider.id])
+ self.assertEqual(
+ GroupMembership.objects.filter(group=self.group, user=self.member).count(), 1
+ )
+ # The regression this guards against: passing the full request list
+ # (including the already-a-member id) would still pass the earlier
+ # assertions above -- only this call proves the notification itself
+ # was narrowed, not just the DB write and the response.
+ notify.assert_called_once()
+ self.assertEqual(notify.call_args.kwargs["user_ids"], [self.outsider.id])
+
def test_service_account_can_remove_member(self) -> None:
response = self._call(
{"delete": "remove_member"},
@@ -482,3 +516,346 @@ def test_descriptors_resolve_and_fields_exist(self) -> None:
f"{resource.kind}.{attr}={field_name!r} is not a field on "
f"{resource.app_label}.{resource.model_name}"
)
+
+
+class ResourceShareNotificationTests(GroupSharingTestBase):
+ """Delivery side (``group_notification_service``): who actually gets mailed.
+
+ The email plugin is mocked, so these pin recipient selection — the live
+ re-read on a grant, the ``revoked_at`` cutoff, org scoping and retained
+ access — not template or transport behavior. The enqueue side is covered in
+ ``test_share_notification_dispatch`` (unit tier, no DB).
+ """
+
+ def setUp(self) -> None:
+ super().setUp()
+ self.service = Mock()
+ patcher = patch(
+ "tenant_account_v2.group_notification_service.notification_plugin",
+ {"service_class": Mock(return_value=self.service)},
+ )
+ patcher.start()
+ self.addCleanup(patcher.stop)
+
+ def _send(
+ self,
+ *,
+ group_ids: list[int],
+ share_action: str = ShareAction.SHARED.value,
+ revoked_at=None,
+ ) -> bool:
+ return send_resource_shared(
+ organization=self.org,
+ group_ids=group_ids,
+ actor_id=self.owner.pk,
+ resource_kind="workflow",
+ resource_id=str(self.workflow.pk),
+ share_action=share_action,
+ revoked_at=revoked_at,
+ )
+
+ def _mailed(self) -> list[tuple[str, list[str]]]:
+ """``(group_name, sorted recipient emails)`` per email sent, in order."""
+ return [
+ (call.kwargs["group_name"], sorted(u.email for u in call.kwargs["shared_to"]))
+ for call in self.service.send_group_resource_shared_notification.call_args_list
+ ]
+
+ def test_grant_mails_current_group_members(self) -> None:
+ set_resource_share_groups(self.workflow, [self.group.id])
+ self._send(group_ids=[self.group.id])
+ self.assertEqual(self._mailed(), [("Team", ["member@example.com"])])
+
+ def test_grant_dropped_when_share_revoked_before_delivery(self) -> None:
+ # The queue can lag; announcing access the group no longer holds would
+ # disclose the resource name and id to members who cannot reach it.
+ set_resource_share_groups(self.workflow, [self.group.id])
+ set_resource_share_groups(self.workflow, [])
+ self._send(group_ids=[self.group.id])
+ self.service.send_group_resource_shared_notification.assert_not_called()
+
+ def test_revoke_mails_members_although_the_share_row_is_gone(self) -> None:
+ # Mirror image of the check above: on a revoke the row is *expected* to
+ # be absent, so the live re-read must not suppress the mail.
+ self._send(group_ids=[self.group.id], share_action=ShareAction.REVOKED.value)
+ self.assertEqual(self._mailed(), [("Team", ["member@example.com"])])
+ kwargs = self.service.send_group_resource_shared_notification.call_args.kwargs
+ self.assertEqual(kwargs["share_action"], "revoked")
+ self.assertEqual(kwargs["resource_type"], "workflow")
+ self.assertEqual(kwargs["resource_name"], "wf-1")
+
+ def test_multi_group_fan_out_keeps_each_group_and_its_own_members_separate(
+ self,
+ ) -> None:
+ # ``self.member`` is in both groups (overlapping); ``self.outsider`` is
+ # only in the second (disjoint). A naive union across groups would
+ # either merge them into one mail or mislabel which group a recipient
+ # actually belongs to -- both are exactly what one email per group
+ # exists to prevent.
+ other_group = OrganizationGroup.objects.create(
+ organization=self.org, name="Ops", created_by=self.owner
+ )
+ GroupMembership.objects.create(group=other_group, user=self.member)
+ GroupMembership.objects.create(group=other_group, user=self.outsider)
+ set_resource_share_groups(self.workflow, [self.group.id, other_group.id])
+
+ self._send(group_ids=[self.group.id, other_group.id])
+
+ self.assertEqual(
+ sorted(self._mailed()),
+ sorted(
+ [
+ ("Team", ["member@example.com"]),
+ ("Ops", ["member@example.com", "outsider@example.com"]),
+ ]
+ ),
+ )
+
+ def test_plugin_skip_result_is_not_a_failure(self) -> None:
+ # The plugin's tri-state None means "skipped" (unconfigured, disabled,
+ # bad input) -- never a reason to ask for redelivery.
+ set_resource_share_groups(self.workflow, [self.group.id])
+ self.service.send_group_resource_shared_notification.return_value = None
+ self.assertTrue(self._send(group_ids=[self.group.id]))
+
+ def test_all_groups_failing_asks_for_redelivery(self) -> None:
+ other_group = OrganizationGroup.objects.create(
+ organization=self.org, name="Ops", created_by=self.owner
+ )
+ GroupMembership.objects.create(group=other_group, user=self.outsider)
+ set_resource_share_groups(self.workflow, [self.group.id, other_group.id])
+ self.service.send_group_resource_shared_notification.return_value = False
+ self.assertFalse(self._send(group_ids=[self.group.id, other_group.id]))
+
+ def test_partial_group_failure_is_not_retried_and_every_group_is_attempted(
+ self,
+ ) -> None:
+ # A retry would re-mail the group that already succeeded -- accept the
+ # partial loss instead. Every group must still be attempted, not just
+ # the ones before the first failure: ThreadPoolExecutor.map's pending
+ # futures must not be cancelled by an early result.
+ other_group = OrganizationGroup.objects.create(
+ organization=self.org, name="Ops", created_by=self.owner
+ )
+ GroupMembership.objects.create(group=other_group, user=self.outsider)
+ set_resource_share_groups(self.workflow, [self.group.id, other_group.id])
+ self.service.send_group_resource_shared_notification.side_effect = [False, True]
+ result = self._send(group_ids=[self.group.id, other_group.id])
+ self.assertTrue(result)
+ self.assertEqual(
+ self.service.send_group_resource_shared_notification.call_count, 2
+ )
+
+ def test_group_from_another_org_is_never_mailed(self) -> None:
+ """The foreign group's member is deliberately also an org-A member.
+
+ Users belong to any number of orgs here, so without that the group has
+ no resolvable recipients and the test passes on an empty list rather
+ than on the org filter -- green even with the filter deleted.
+ """
+ other_org = Organization.objects.create(
+ name="org-b", display_name="Org B", organization_id="org-b"
+ )
+ foreign_group = OrganizationGroup.objects.create(
+ organization=other_org, name="Foreign", created_by=self.owner
+ )
+ dual = _make_user("dual@example.com")
+ OrganizationMember.objects.create(organization=self.org, user=dual, role="user")
+ OrganizationMember.objects.create(organization=other_org, user=dual, role="user")
+ GroupMembership.objects.create(group=foreign_group, user=dual)
+
+ for action in (ShareAction.SHARED.value, ShareAction.REVOKED.value):
+ self._send(group_ids=[foreign_group.id], share_action=action)
+ self.assertEqual(self._mailed(), [])
+
+ def test_group_member_outside_the_org_is_never_mailed(self) -> None:
+ """A group row that outlives the org membership must not produce mail.
+
+ Pins the outcome, not the layer: ``OrganizationMember``'s default
+ manager is org-scoped, so deleting the explicit filter in
+ ``_live_member_users`` changes nothing. ``OrganizationGroup`` has no
+ such manager, which is why the group-level filter IS pinnable -- see
+ ``test_group_from_another_org_is_never_mailed``.
+ """
+ other_org = Organization.objects.create(
+ name="org-d", display_name="Org D", organization_id="org-d"
+ )
+ # A member of ANOTHER org, not of no org: a user with no membership row
+ # at all is excluded by the table rather than by the org clause, which
+ # would leave this test green with the filter deleted.
+ stranger = _make_user("stranger@example.com")
+ OrganizationMember.objects.create(
+ organization=other_org, user=stranger, role="user"
+ )
+ GroupMembership.objects.create(group=self.group, user=stranger)
+ set_resource_share_groups(self.workflow, [self.group.id])
+ self._send(group_ids=[self.group.id])
+ self.assertEqual(self._mailed(), [("Team", ["member@example.com"])])
+
+ def test_actor_outside_the_org_is_not_resolved(self) -> None:
+ """The actor's name and email render into the outgoing mail, so an
+ actor from another org must not resolve. Outcome-level, like the
+ recipient case above: the org-scoped manager enforces it either way.
+ """
+ other_org = Organization.objects.create(
+ name="org-e", display_name="Org E", organization_id="org-e"
+ )
+ # Again a member of another org rather than of none, so the org clause
+ # is the only thing that can exclude them.
+ foreign_actor = _make_user("foreign-actor@example.com")
+ OrganizationMember.objects.create(
+ organization=other_org, user=foreign_actor, role="user"
+ )
+ # The grant direction drops a group with no live share row, which would
+ # stop the mail before the actor is ever resolved.
+ set_resource_share_groups(self.workflow, [self.group.id])
+ send_resource_shared(
+ organization=self.org,
+ group_ids=[self.group.id],
+ actor_id=foreign_actor.pk,
+ resource_kind="workflow",
+ resource_id=str(self.workflow.pk),
+ share_action=ShareAction.SHARED.value,
+ revoked_at=None,
+ )
+ self.service.send_group_resource_shared_notification.assert_not_called()
+
+ def test_resource_from_another_org_is_not_resolved(self) -> None:
+ """A resource id belonging to another org must not resolve.
+
+ For ``Workflow`` the org-scoped manager already enforces this, so this
+ pins the outcome rather than the explicit filter. That filter exists
+ for ``AgenticProject``, whose manager deliberately spans orgs -- a
+ cloud-only model, so the case it guards cannot be exercised here.
+ """
+ other_org = Organization.objects.create(
+ name="org-c", display_name="Org C", organization_id="org-c"
+ )
+ foreign_wf = Workflow.objects.create(
+ workflow_name="wf-other", organization=other_org, created_by=self.owner
+ )
+ with self.assertRaises(ResourceNotFoundError):
+ send_resource_shared(
+ organization=self.org,
+ group_ids=[self.group.id],
+ actor_id=self.owner.pk,
+ resource_kind="workflow",
+ resource_id=str(foreign_wf.pk),
+ share_action=ShareAction.SHARED.value,
+ revoked_at=None,
+ )
+ self.service.send_group_resource_shared_notification.assert_not_called()
+
+ def test_revoke_on_an_org_shared_resource_mails_nobody(self) -> None:
+ """Nobody lost access, so nobody is told.
+
+ The short-circuit that skips hydrating every org member to reach this
+ answer is an optimisation, not a behaviour change -- removing it leaves
+ this assertion green. Only a query count would pin that half.
+ """
+ self.workflow.shared_to_org = True
+ self.workflow.save(update_fields=["shared_to_org"])
+ self._send(group_ids=[self.group.id], share_action=ShareAction.REVOKED.value)
+ self.assertEqual(self._mailed(), [])
+
+ def test_revoke_does_not_tell_an_owner_they_lost_access(self) -> None:
+ """Owners sit outside ``compute_effective_members``, so they have to be
+ added back explicitly or an owner inside a revoked group is mailed a
+ false removal notice.
+ """
+ GroupMembership.objects.create(group=self.group, user=self.owner)
+ self._send(group_ids=[self.group.id], share_action=ShareAction.REVOKED.value)
+ self.assertEqual(self._mailed(), [("Team", ["member@example.com"])])
+
+ def test_revoke_does_not_tell_an_org_admin_they_lost_access(self) -> None:
+ """An admin reaches every resource in the org via ``for_user``, so a
+ revoke takes nothing from them.
+
+ The admin ROLE STRING differs between the OSS and auth0 auth plugins,
+ so the predicate is patched rather than resolved for real.
+ """
+ GroupMembership.objects.create(group=self.group, user=self.admin)
+ with patch(
+ "account_v2.authentication_controller.AuthenticationController"
+ ".is_admin_by_role",
+ side_effect=lambda role: role == "admin",
+ ):
+ self._send(group_ids=[self.group.id], share_action=ShareAction.REVOKED.value)
+ self.assertEqual(self._mailed(), [("Team", ["member@example.com"])])
+
+ def test_membership_removal_is_mailed_as_a_removal(self) -> None:
+ """The ADDED direction was the only one exercised, so hardcoding the
+ action passed every test while telling removed users they were added.
+ """
+ send_membership_changed(
+ organization=self.org,
+ group_id=self.group.id,
+ actor_id=self.owner.pk,
+ membership_action=MembershipAction.REMOVED.value,
+ user_ids=[self.member.pk],
+ )
+ kwargs = self.service.send_group_membership_notification.call_args.kwargs
+ self.assertEqual(kwargs["membership_action"], "removed")
+
+ def test_membership_recipients_are_revalidated_against_the_org(self) -> None:
+ """Leaving a group does not remove someone from the org, and leaving the
+ org does not delete their group rows -- so the recipient list is filtered
+ on OrganizationMember rather than taken from the payload.
+ """
+ stranger = _make_user("ex@example.com")
+ send_membership_changed(
+ organization=self.org,
+ group_id=self.group.id,
+ actor_id=self.owner.pk,
+ membership_action=MembershipAction.REMOVED.value,
+ user_ids=[self.member.pk, stranger.pk],
+ )
+ kwargs = self.service.send_group_membership_notification.call_args.kwargs
+ self.assertEqual([u.email for u in kwargs["recipients"]], ["member@example.com"])
+
+ def test_revoke_skips_members_who_joined_after_the_cutoff(self) -> None:
+ revoked_at = timezone.now()
+ latecomer = GroupMembership.objects.create(group=self.group, user=self.outsider)
+ # ``created_at`` is auto-set on save, so move it past the cutoff directly.
+ GroupMembership.objects.filter(pk=latecomer.pk).update(
+ created_at=revoked_at + timedelta(minutes=1)
+ )
+ self._send(
+ group_ids=[self.group.id],
+ share_action=ShareAction.REVOKED.value,
+ revoked_at=revoked_at,
+ )
+ # ``outsider`` never held access through this group, so no revoke notice.
+ self.assertEqual(self._mailed(), [("Team", ["member@example.com"])])
+
+ def test_revoke_skips_members_who_keep_access_another_way(self) -> None:
+ _add_viewers(self.workflow, self.member)
+ self._send(group_ids=[self.group.id], share_action=ShareAction.REVOKED.value)
+ # Nothing was lost — a direct VIEWER row still reaches the resource.
+ self.service.send_group_resource_shared_notification.assert_not_called()
+
+ def test_revoke_skips_members_who_keep_access_via_another_group(self) -> None:
+ # ``self.member`` is in both groups; only ``self.group`` is revoked.
+ other_group = OrganizationGroup.objects.create(
+ organization=self.org, name="Ops", created_by=self.owner
+ )
+ GroupMembership.objects.create(group=other_group, user=self.member)
+ set_resource_share_groups(self.workflow, [other_group.id])
+ self._send(group_ids=[self.group.id], share_action=ShareAction.REVOKED.value)
+ # Still reaches the resource through "Ops" -- nothing was lost.
+ self.service.send_group_resource_shared_notification.assert_not_called()
+
+ def test_membership_change_mails_only_the_changed_users(self) -> None:
+ send_membership_changed(
+ organization=self.org,
+ group_id=self.group.id,
+ actor_id=self.owner.pk,
+ membership_action=MembershipAction.ADDED.value,
+ user_ids=[self.outsider.pk],
+ )
+ kwargs = self.service.send_group_membership_notification.call_args.kwargs
+ self.assertEqual(kwargs["group_name"], "Team")
+ self.assertEqual(kwargs["membership_action"], "added")
+ self.assertEqual(
+ [u.email for u in kwargs["recipients"]], ["outsider@example.com"]
+ )
diff --git a/backend/workflow_manager/workflow_v2/views.py b/backend/workflow_manager/workflow_v2/views.py
index 845b254177..938ed6c534 100644
--- a/backend/workflow_manager/workflow_v2/views.py
+++ b/backend/workflow_manager/workflow_v2/views.py
@@ -171,51 +171,6 @@ def perform_create(self, serializer: WorkflowSerializer) -> Workflow:
raise WorkflowGenerationError
return workflow
- def partial_update(self, request: Request, *args: Any, **kwargs: Any) -> Response:
- """Override partial_update to handle sharing notifications."""
- workflow = self.get_object()
- before = self.snapshot_share_axes(workflow)
-
- response = super().partial_update(request, *args, **kwargs)
- if response.status_code == 200 and notification_plugin:
- self._notify_shared_users(workflow, before, request.data, request.user)
- return response
-
- def _notify_shared_users(
- self,
- workflow: Workflow,
- 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(workflow, 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.WORKFLOW.value,
- resource_name=workflow.workflow_name,
- resource_id=str(workflow.id),
- shared_by=actor,
- shared_to=list(users_diff.added),
- resource_instance=workflow,
- )
- logger.info(
- "Sent sharing notifications for workflow %s to %d users",
- workflow.id,
- len(users_diff.added),
- )
- except Exception as e:
- logger.exception(
- "Failed to send sharing notification, continuing update though: %s",
- str(e),
- )
-
def get_execution(self, request: Request, pk: str) -> Response:
execution = WorkflowHelper.get_current_execution(pk)
return Response(make_execution_response(execution), status=status.HTTP_200_OK)
diff --git a/frontend/src/components/deployments/api-deployment/ApiDeployment.jsx b/frontend/src/components/deployments/api-deployment/ApiDeployment.jsx
index 21ff2a44c3..3cae374d22 100644
--- a/frontend/src/components/deployments/api-deployment/ApiDeployment.jsx
+++ b/frontend/src/components/deployments/api-deployment/ApiDeployment.jsx
@@ -65,8 +65,7 @@ function ApiDeployment() {
coOwnerAllUsers,
coOwnerResourceId,
handleCoOwner: handleCoOwnerAction,
- onAddCoOwner,
- onRemoveCoOwner,
+ onApplyCoOwners,
} = useCoOwnerManagement({
service: apiDeploymentsApiService,
setAlertDetails,
@@ -410,10 +409,8 @@ function ApiDeployment() {
resourceType="API Deployment"
allUsers={coOwnerAllUsers}
coOwners={coOwnerData.coOwners}
- createdBy={coOwnerData.createdBy}
loading={coOwnerLoading}
- onAddCoOwner={onAddCoOwner}
- onRemoveCoOwner={onRemoveCoOwner}
+ onApplyCoOwners={onApplyCoOwners}
/>
);
diff --git a/frontend/src/components/pipelines-or-deployments/notification-modal/CreateNotification.jsx b/frontend/src/components/pipelines-or-deployments/notification-modal/CreateNotification.jsx
index 5ed4382b1a..44c6df5a16 100644
--- a/frontend/src/components/pipelines-or-deployments/notification-modal/CreateNotification.jsx
+++ b/frontend/src/components/pipelines-or-deployments/notification-modal/CreateNotification.jsx
@@ -1,5 +1,5 @@
import PropTypes from "prop-types";
-import { useEffect, useState } from "react";
+import { useState } from "react";
import { Button } from "@/components/ui/shims/antd-button";
import { Form } from "@/components/ui/shims/antd-form";
import { Checkbox, Input, Select } from "@/components/ui/shims/antd-inputs";
@@ -68,23 +68,11 @@ function CreateNotification({
editDetails,
}) {
const [form] = Form.useForm();
- const [formDetails, setFormDetails] = useState(DEFAULT_FORM_DETAILS);
+ // Lazy init: the Form shim seeds initialValues only on first mount.
+ const [formDetails, setFormDetails] = useState(
+ () => editDetails ?? DEFAULT_FORM_DETAILS,
+ );
const [backendErrors, setBackendErrors] = useState(null);
- const [resetForm, setResetForm] = useState(false);
-
- useEffect(() => {
- if (editDetails) {
- setFormDetails(editDetails);
- setResetForm(true);
- }
- }, [editDetails]);
-
- useEffect(() => {
- if (resetForm) {
- form.resetFields();
- setResetForm(false);
- }
- }, [formDetails]);
const handleInputChange = (changedValues, allValues) => {
const nextValues = { ...formDetails, ...allValues };
diff --git a/frontend/src/components/pipelines-or-deployments/pipelines/Pipelines.jsx b/frontend/src/components/pipelines-or-deployments/pipelines/Pipelines.jsx
index edde00911e..52c38e0f8f 100644
--- a/frontend/src/components/pipelines-or-deployments/pipelines/Pipelines.jsx
+++ b/frontend/src/components/pipelines-or-deployments/pipelines/Pipelines.jsx
@@ -69,8 +69,7 @@ function Pipelines({ type }) {
coOwnerAllUsers,
coOwnerResourceId,
handleCoOwner: handleCoOwnerAction,
- onAddCoOwner,
- onRemoveCoOwner,
+ onApplyCoOwners,
} = useCoOwnerManagement({
service: pipelineApiService,
setAlertDetails,
@@ -487,10 +486,8 @@ function Pipelines({ type }) {
resourceType="Pipeline"
allUsers={coOwnerAllUsers}
coOwners={coOwnerData.coOwners}
- createdBy={coOwnerData.createdBy}
loading={coOwnerLoading}
- onAddCoOwner={onAddCoOwner}
- onRemoveCoOwner={onRemoveCoOwner}
+ onApplyCoOwners={onApplyCoOwners}
/>
)}
diff --git a/frontend/src/components/widgets/co-owner-management/CoOwnerManagement.css b/frontend/src/components/widgets/co-owner-management/CoOwnerManagement.css
index c7698eea62..75eb050515 100644
--- a/frontend/src/components/widgets/co-owner-management/CoOwnerManagement.css
+++ b/frontend/src/components/widgets/co-owner-management/CoOwnerManagement.css
@@ -3,10 +3,6 @@
margin-bottom: 16px;
}
-.co-owner-creator-tag {
- margin-left: 8px;
-}
-
.co-owner-modal .shared-user-avatar {
background-color: #00a6ed;
margin-right: 15px;
diff --git a/frontend/src/components/widgets/co-owner-management/CoOwnerManagement.jsx b/frontend/src/components/widgets/co-owner-management/CoOwnerManagement.jsx
index 9ca010329b..280f8cc030 100644
--- a/frontend/src/components/widgets/co-owner-management/CoOwnerManagement.jsx
+++ b/frontend/src/components/widgets/co-owner-management/CoOwnerManagement.jsx
@@ -19,82 +19,90 @@ function CoOwnerManagement({
allUsers,
coOwners,
loading,
- onAddCoOwner,
- onRemoveCoOwner,
+ onApplyCoOwners,
}) {
- const [pendingAdds, setPendingAdds] = useState([]);
- const [removingUserId, setRemovingUserId] = useState(null);
+ // Staged roster. Adds and removals both edit this list only — nothing reaches
+ // the API until Apply, the same contract as the share modal.
+ const [selectedOwners, setSelectedOwners] = useState([]);
const [applying, setApplying] = useState(false);
- const ownersList = coOwners || [];
- const totalOwners = ownersList.length;
-
- // Exclude both existing co-owners and pending adds from dropdown
- const availableUsers = useMemo(() => {
- const coOwnerIds = new Set((coOwners || []).map((u) => u?.id?.toString()));
- const pendingIds = new Set(pendingAdds.map((u) => u?.id?.toString()));
- return (allUsers || []).filter(
- (user) =>
- !coOwnerIds.has(user?.id?.toString()) &&
- !pendingIds.has(user?.id?.toString()),
- );
- }, [allUsers, coOwners, pendingAdds]);
+ const ownersList = useMemo(() => coOwners || [], [coOwners]);
+
+ // Re-seed whenever the server roster changes: on open, on resource switch, and
+ // after an Apply. Doubles as the reset — most hosts leave this modal mounted,
+ // and the hook can close it without ``handleCancel`` (404 / fetch-error), so
+ // staged edits must not leak into the next resource.
+ //
+ // Done during render, not in an effect: an effect commits after the one that
+ // reveals the new roster, so the first frame of a new resource would render
+ // the previous resource's staged list (and enable Apply on that diff).
+ const [seededFrom, setSeededFrom] = useState(null);
+ if (seededFrom !== ownersList) {
+ setSeededFrom(ownersList);
+ setSelectedOwners(ownersList);
+ }
+
+ const selectedIds = useMemo(
+ () => new Set(selectedOwners.map((u) => u?.id?.toString())),
+ [selectedOwners],
+ );
+
+ const availableUsers = useMemo(
+ () => (allUsers || []).filter((u) => !selectedIds.has(u?.id?.toString())),
+ [allUsers, selectedIds],
+ );
+
+ const { addUsers, removeUsers } = useMemo(() => {
+ const ownerIds = new Set(ownersList.map((u) => u?.id?.toString()));
+ return {
+ addUsers: selectedOwners.filter((u) => !ownerIds.has(u?.id?.toString())),
+ removeUsers: ownersList.filter(
+ (u) => !selectedIds.has(u?.id?.toString()),
+ ),
+ };
+ }, [ownersList, selectedOwners, selectedIds]);
+
+ const hasChanges = addUsers.length > 0 || removeUsers.length > 0;
const handleSelect = (userId) => {
const user = (allUsers || []).find(
(u) => u?.id?.toString() === userId?.toString(),
);
if (user) {
- setPendingAdds((prev) => [...prev, user]);
+ setSelectedOwners((prev) => [...prev, user]);
}
};
- const handleRemovePending = (userId) => {
- setPendingAdds((prev) =>
+ const handleRemove = (userId) => {
+ setSelectedOwners((prev) =>
prev.filter((u) => u?.id?.toString() !== userId?.toString()),
);
};
- const handleRemoveExisting = async (userId) => {
- setRemovingUserId(userId);
- try {
- await onRemoveCoOwner(resourceId, userId);
- } finally {
- setRemovingUserId(null);
- }
- };
-
const handleApply = async () => {
- if (pendingAdds.length === 0) return;
- const usersToAdd = [...pendingAdds];
+ if (!hasChanges) {
+ return;
+ }
setApplying(true);
try {
- const userIds = usersToAdd.map((user) => user.id);
- await onAddCoOwner(resourceId, userIds);
+ // Close only on a clean apply. A partial failure keeps the modal open on
+ // the refreshed server roster, with the alert naming who was rejected.
+ if (await onApplyCoOwners(resourceId, { addUsers, removeUsers })) {
+ setOpen(false);
+ }
} finally {
- setPendingAdds([]);
setApplying(false);
}
};
const handleCancel = () => {
- setPendingAdds([]);
+ setSelectedOwners(ownersList);
setOpen(false);
};
const filterOption = (input, option) =>
(option?.label ?? "").toLowerCase().includes(input.toLowerCase());
- const combinedList = [
- ...ownersList,
- ...pendingAdds.filter(
- (pending) =>
- !ownersList.some(
- (owner) => owner?.id?.toString() === pending?.id?.toString(),
- ),
- ),
- ];
-
return (
Co-Owners
- {combinedList.length > 0 ? (
+ {selectedOwners.length > 0 ? (
{
- const isPending = pendingAdds.some(
- (u) => u?.id?.toString() === item?.id?.toString(),
- );
- return (
- }
- onClick={() => handleRemovePending(item?.id)}
- aria-label={`Remove pending co-owner ${item?.email}`}
+ dataSource={selectedOwners}
+ renderItem={(item) => (
+ 1 && (
+