From 798438d7a64826976e8d543f5a011e780a71f631 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Fri, 31 Jul 2026 20:41:31 +0530 Subject: [PATCH 01/10] UN-3494 [FEAT] Email group members on resource share and group membership changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sharing a resource with a group gave its members access silently, and adding or removing someone from a group told nobody. Both now send email. - share_notifications.py holds the feature flag, the two task names and the two enqueue hooks. Dispatch uses the same resolve_transport branch the execution path uses: the PG queue where pg_queue_enabled is on for the org, Celery otherwise. - One hook in ResourceShareManagementMixin.share covers all 7 resource types plus cloud agentic, including service-account shares — every group share funnels through it and shared_groups has no PATCH path. No on_commit needed: _commit's transaction has closed by the time the view resumes, so the diff reads committed state. - Group membership hooks on the add and remove actions. The add serializer already subtracts existing members, so nobody is mailed twice. - Internal endpoints under /internal/v1/group-notification/ do the work the worker cannot: group expansion, OrganizationMember re-validation (this is where the offboarding race closes), resource lookup via ShareableResource, and the kind -> ResourceType mapping, which is not 1:1 — pipelines split on pipeline_type and adapters four ways on adapter_type. - Two worker tasks that only POST to that endpoint, since workers/ has no Django. They raise on failure, unlike _mark_buffer_outcome which has a reaper behind it, and retry transient 5xx in-task because a raise is terminal on the Celery transport. - The whole feature is gated on Flipt group_sharing_notifications_enabled and fails closed: a blind Flipt, a missing org, or any dispatch error means no notification, never a broken share. - worker-pg-notification compose service so the PG arm is not a black hole. Membership changes with no actor (the org-removal cascade, Django admin, group deletion) do not notify — SharingNotificationService requires an actor. Co-Authored-By: Claude Opus 5 --- backend/backend/internal_base_urls.py | 6 + backend/permissions/resource_share_views.py | 16 ++ .../group_notification_service.py | 237 ++++++++++++++++++ backend/tenant_account_v2/group_views.py | 18 ++ backend/tenant_account_v2/internal_urls.py | 20 ++ backend/tenant_account_v2/internal_views.py | 91 +++++++ .../tenant_account_v2/share_notifications.py | 210 ++++++++++++++++ .../tenant_account_v2/shareable_resources.py | 24 ++ docker/docker-compose.yaml | 36 +++ workers/notification/tasks.py | 99 ++++++++ 10 files changed, 757 insertions(+) create mode 100644 backend/tenant_account_v2/group_notification_service.py create mode 100644 backend/tenant_account_v2/internal_urls.py create mode 100644 backend/tenant_account_v2/internal_views.py create mode 100644 backend/tenant_account_v2/share_notifications.py diff --git a/backend/backend/internal_base_urls.py b/backend/backend/internal_base_urls.py index 0354a691ae..e89c5519f1 100644 --- a/backend/backend/internal_base_urls.py +++ b/backend/backend/internal_base_urls.py @@ -269,4 +269,10 @@ def test_middleware_debug(request): include("prompt_studio.prompt_studio_core_v2.internal_urls"), name="prompt_studio_internal", ), + # Group-sharing email notification APIs + path( + "v1/group-notification/", + include("tenant_account_v2.internal_urls"), + name="group_notification_internal", + ), ] diff --git a/backend/permissions/resource_share_views.py b/backend/permissions/resource_share_views.py index 1227562531..f688569115 100644 --- a/backend/permissions/resource_share_views.py +++ b/backend/permissions/resource_share_views.py @@ -91,13 +91,29 @@ 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_shared_with_group, + ) from tenant_account_v2.sharing_helpers import ShareAuthorizationService resource = self.get_object() # type: ignore[attr-defined] desired = _extract_desired_share_state(request.data) + # Only the groups axis is diffed: it is the one that notifies, and + # snapshotting ``shared_users`` too would fetch every viewer twice for + # nothing. Reads go through ``ResourceGroupShare``, so no refresh is + # needed between the two. + groups_before = self._read_axis(resource, "shared_groups") ShareAuthorizationService.authorize_and_commit( actor=request.user, resource=resource, desired=desired ) + # ``_commit`` is the only atomic block on this path, so it has already + # committed — the diff reads persisted state and can never announce a + # share that rolled back. + notify_resource_shared_with_group( + resource=resource, + groups=self._read_axis(resource, "shared_groups") - groups_before, + actor=request.user, + ) return Response(status=status.HTTP_200_OK) @action(detail=True, methods=["get"], url_path="effective-members") 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..53f2bc7f82 --- /dev/null +++ b/backend/tenant_account_v2/group_notification_service.py @@ -0,0 +1,237 @@ +"""Send-side logic for group-sharing email notifications (UN-3494 / mfbt UNS-848). + +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 typing import TYPE_CHECKING, Any + +from account_v2.models import Organization, User +from django.apps import apps +from django.db.models import QuerySet +from plugins import get_plugin + +from tenant_account_v2.models import OrganizationGroup, OrganizationMember +from tenant_account_v2.share_notifications import MembershipAction +from tenant_account_v2.shareable_resources import ShareableResource, descriptor_for_kind + +if TYPE_CHECKING: + from collections.abc import Iterable + +logger = logging.getLogger(__name__) + +notification_plugin = get_plugin("notification") + +# OSS ``ShareableResource.kind`` → the email plugin's ``ResourceType`` value. +# Deliberately plain strings: OSS must not import a cloud-only enum. Not a 1:1 +# rename — pipelines and adapters resolve from the instance below. +_STATIC_RESOURCE_TYPES = { + "workflow": "workflow", + "api_deployment": "api", + "connector_instance": "connector", + "custom_tool": "text_extractor", + "agentic_project": "agentic_project", +} +_ADAPTER_RESOURCE_TYPES = { + "LLM": "llm", + "EMBEDDING": "embedding", + "VECTOR_DB": "vector_db", + "X2TEXT": "x2text", +} +# Only ETL/TASK pipelines map to a notification resource type; the plugin +# compares against these exact (uppercase) values. +_PIPELINE_RESOURCE_TYPES = frozenset({"ETL", "TASK"}) + + +class ResourceNotFoundError(Exception): + """The shared resource no longer exists, or is not in the given org.""" + + +def send_resource_shared( + *, + organization: Organization, + group_ids: Iterable[int], + actor_id: int, + resource_kind: str, + resource_id: str, +) -> None: + """Mail every current member of each group that a resource was shared. + + One email per group, so ``group_name`` in the template is always the group + the recipient actually belongs to. + """ + service = _service() + if service is None: + return + actor = _get_user(actor_id) + resource, resource_name, resource_type = _load_resource( + organization, resource_kind, resource_id + ) + if actor is None or resource_type is None: + logger.info( + "group-notification: skipping resource share for %s/%s " + "(actor_found=%s resource_type=%s)", + resource_kind, + resource_id, + actor is not None, + resource_type, + ) + return + for group in _groups_in_org(organization, group_ids): + recipients = _live_member_users( + organization, group.memberships.values_list("user_id", flat=True) + ) + logger.info( + "group-notification: task=%s group_id=%s recipient_count=%d", + "notify_resource_shared_with_group", + group.pk, + len(recipients), + ) + if not recipients: + continue + service.send_group_resource_shared_notification( + resource_type=resource_type, + resource_name=resource_name, + resource_id=str(resource.pk), + group_name=group.name, + shared_by=actor, + shared_to=recipients, + resource_instance=resource, + ) + + +def send_membership_changed( + *, + organization: Organization, + group_id: int, + actor_id: int, + membership_action: str, + user_ids: Iterable[int], +) -> None: + """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. + """ + service = _service() + if service is None: + return + actor = _get_user(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 + 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 + service.send_group_membership_notification( + group_name=group.name, + membership_action=MembershipAction(membership_action).value, + recipients=recipients, + actor=actor, + organization=organization, + ) + + +def _service() -> Any | None: + """The cloud email service, or ``None`` when the plugin is absent (OSS).""" + if not notification_plugin: + logger.debug("group-notification: notification plugin unavailable, skipping") + return None + return notification_plugin["service_class"]() + + +def _get_user(user_id: int) -> User | None: + return User.objects.filter(pk=user_id).first() + + +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``. + """ + memberships = OrganizationMember.objects.filter( + organization=organization, user_id__in=list(user_ids) + ).select_related("user") + return [ + m.user + for m in memberships + if not getattr(m.user, "is_service_account", False) and m.user.email + ] + + +def _load_resource( + organization: Organization, kind: str, resource_id: str +) -> tuple[Any, str, str | None]: + """Resolve the shared resource to ``(instance, display name, plugin type)``. + + 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") + name = getattr(resource, descriptor.name_field, "") or "" + return 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": + pipeline_type = getattr(resource, "pipeline_type", None) + return pipeline_type if pipeline_type in _PIPELINE_RESOURCE_TYPES else None + if descriptor.kind == "adapter_instance": + # Unknown adapter types fall back to ``llm``, matching the co-owner + # path's override — an OCR adapter shared with a group should not + # silently send nothing when sharing it with a co-owner mails fine. + return _ADAPTER_RESOURCE_TYPES.get(str(resource.adapter_type or ""), "llm") + 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..d5013b81b8 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__) @@ -155,6 +159,14 @@ def members(self, request: Request, pk: str | None = None) -> Response: [GroupMembership(group=group, user_id=uid) for uid in user_ids_to_add], ignore_conflicts=True, ) + # The serializer already subtracts existing members, so nobody gets a + # second "you've been added" mail for a group they were already in. + notify_group_membership_changed( + group=group, + action=MembershipAction.ADDED, + user_ids=user_ids_to_add, + actor=request.user, + ) return Response( {"added_user_ids": user_ids_to_add}, status=status.HTTP_201_CREATED, @@ -178,6 +190,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..fb707df14e --- /dev/null +++ b/backend/tenant_account_v2/internal_views.py @@ -0,0 +1,91 @@ +"""Internal API views for group-sharing email notifications (UN-3494 / UNS-848). + +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. + +Failure contract: **any** unhandled problem must surface as non-2xx so the +queue redelivers. The one deliberate exception is a resource that no longer +exists, which returns 200 — retrying that can only fail again. +""" + +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 + +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() + + +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 a resource was just shared with.""" + + def post(self, request: Request) -> Response: + serializer = ResourceSharedWithGroupSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + data = serializer.validated_data + try: + 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) + 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) + send_membership_changed( + organization=self._organization(), **serializer.validated_data + ) + return Response({"status": "success"}, status=status.HTTP_200_OK) diff --git a/backend/tenant_account_v2/share_notifications.py b/backend/tenant_account_v2/share_notifications.py new file mode 100644 index 0000000000..c4191a8a80 --- /dev/null +++ b/backend/tenant_account_v2/share_notifications.py @@ -0,0 +1,210 @@ +"""Enqueue hooks for group-sharing email notifications (UN-3494 / mfbt UNS-848). + +Two events earn a group's members an email: a resource shared with 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 resolved per-org by the same +``resolve_transport`` gate the execution path uses — the PG queue where that is +enabled, Celery otherwise. + +The whole feature sits behind its own Flipt flag and fails closed everywhere: a +blind Flipt, a missing org, or any dispatch error means no notification, never +a broken share. +""" + +from __future__ import annotations + +import logging +import os +from collections.abc import Iterable +from enum import StrEnum +from typing import TYPE_CHECKING, Any + +from tenant_account_v2.shareable_resources import kind_for_instance +from unstract.core.data_models import is_pg_transport +from unstract.flags.feature_flag import check_feature_flag_status + +if TYPE_CHECKING: + from account_v2.models import User + + from tenant_account_v2.models import OrganizationGroup + +logger = logging.getLogger(__name__) + +# Rollout flag for the whole feature. Sibling of ``pg_queue.flags`` — kept in +# one place so a grep on the constant finds every gate. +GROUP_NOTIFICATION_FLAG_KEY = "group_sharing_notifications_enabled" + +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" + + +def notify_resource_shared_with_group( + *, resource: Any, groups: Iterable[OrganizationGroup], actor: User +) -> None: + """Queue "a resource was shared with your group" mail for newly added groups. + + 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. + """ + group_ids = sorted(group.pk for group in groups) + if not group_ids: + return + organization_id = _organization_slug(resource) + kind = kind_for_instance(resource) + if not organization_id or kind is None or not _feature_enabled(organization_id): + return + _dispatch_quietly( + task_name=NOTIFY_RESOURCE_SHARED_TASK, + kwargs={ + "group_ids": group_ids, + "actor_id": actor.pk, + "resource_kind": kind, + "resource_id": str(resource.pk), + "organization_id": organization_id, + }, + organization_id=organization_id, + entity_id=str(resource.pk), + ) + + +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 or not _feature_enabled(organization_id): + return + _dispatch_quietly( + task_name=NOTIFY_MEMBERSHIP_CHANGED_TASK, + kwargs={ + "group_id": group.pk, + "actor_id": actor.pk, + "membership_action": str(action), + "user_ids": recipients, + "organization_id": organization_id, + }, + organization_id=organization_id, + entity_id=str(group.pk), + ) + + +def _feature_enabled(organization_id: str) -> bool: + """Whether group-sharing notifications are on for this org. Fails closed.""" + # Parse exactly as FliptClient does (``.lower()``, no ``.strip()``) so the + # two can never disagree on a value like " true". + if os.environ.get("FLIPT_SERVICE_AVAILABLE", "false").lower() != "true": + return False + try: + return bool( + check_feature_flag_status( + flag_key=GROUP_NOTIFICATION_FLAG_KEY, + entity_id=organization_id, + context={"organization_id": organization_id}, + ) + ) + except Exception: + logger.warning( + "group-notification: Flipt evaluation failed for org %s; skipping", + organization_id, + exc_info=True, + ) + return False + + +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 ``resolve_transport`` expects. + """ + organization = getattr(obj, "organization", None) + return getattr(organization, "organization_id", None) + + +def _dispatch_quietly( + *, + task_name: str, + kwargs: dict[str, Any], + organization_id: str, + entity_id: str, +) -> None: + """Dispatch on the resolved transport; 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. + """ + try: + _dispatch( + task_name=task_name, + kwargs=kwargs, + organization_id=organization_id, + entity_id=entity_id, + ) + except Exception: + logger.exception( + "group-notification: failed to dispatch %s for org %s", + task_name, + organization_id, + ) + + +def _dispatch( + *, + task_name: str, + kwargs: dict[str, Any], + organization_id: str, + entity_id: str, +) -> None: + # Lazy imports — ``backend.celery_service`` and ``pg_queue`` are heavier + # than this leaf module and importing them at load time risks a cycle + # during Django app loading. + from pg_queue.producer import enqueue_task + from workflow_manager.workflow_v2.transport import resolve_transport + + from backend.celery_service import app as celery_app + + transport = resolve_transport(execution_id=entity_id, organization_id=organization_id) + if is_pg_transport(transport): + 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, + ) + return + celery_app.send_task(task_name, kwargs=kwargs, queue=NOTIFICATION_QUEUE) + logger.info("group-notification: %s dispatched on Celery", task_name) diff --git a/backend/tenant_account_v2/shareable_resources.py b/backend/tenant_account_v2/shareable_resources.py index f528e2959f..2d55e0de0d 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) @@ -49,3 +50,26 @@ class ShareableResource: "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/docker/docker-compose.yaml b/docker/docker-compose.yaml index 9a8db90afe..d74dc2bf43 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -827,6 +827,42 @@ services: profiles: - pg-queue + # Notification consumer — webhook POSTs and the group-share emails (UN-3494). + # Without this, a notification enqueued on PG is durably stored and never run. + # Every task here is one short outbound HTTP call, so it stays light. + worker-pg-notification: + image: unstract/worker-unified:${VERSION} + container_name: unstract-worker-pg-notification + restart: unless-stopped + command: ["pg-queue-consumer"] + ports: + - "8101:8090" + env_file: + - ../workers/.env + - ./essentials.env + depends_on: + - db + - redis + environment: + - ENVIRONMENT=development + - APPLICATION_NAME=unstract-worker-pg-notification + - WORKER_BARRIER_BACKEND=pg + - WORKER_PG_QUEUE_CONSUMER_WORKER_TYPE=notification + - WORKER_PG_QUEUE_CONSUMER_QUEUE=notifications + - WORKER_PG_QUEUE_CONSUMER_HEALTH_PORT=8090 + - WORKER_PG_QUEUE_CONSUMER_CONCURRENCY=${PG_NOTIFICATION_CONCURRENCY:-4} + # One internal-API call (30s timeout) plus a SendGrid batch for the + # largest group, with headroom. Health-stale sits at or above it. + - WORKER_PG_QUEUE_CONSUMER_VT_SECONDS=${PG_NOTIFICATION_VT_SECONDS:-120} + - WORKER_PG_QUEUE_CONSUMER_HEALTH_STALE_SECONDS=${PG_NOTIFICATION_HEALTH_STALE_SECONDS:-180} + labels: + - traefik.enable=false + volumes: + - ./workflow_data:/data + - ${TOOL_REGISTRY_CONFIG_SRC_PATH}:/data/tool_registry_config + profiles: + - pg-queue + # Reaper / orchestrator — leader-elected loop. Run exactly ONE instance (it # elects a single leader via pg_orchestrator_lock; extra replicas idle as # standby). Besides barrier-orphan recovery it runs the PG scheduler tick diff --git a/workers/notification/tasks.py b/workers/notification/tasks.py index 41d76f5cf9..f6e3d1ee9c 100644 --- a/workers/notification/tasks.py +++ b/workers/notification/tasks.py @@ -6,6 +6,7 @@ """ import os +import time from typing import Any import httpx @@ -467,6 +468,104 @@ def priority_notification(notification_type: str, **kwargs: Any) -> dict[str, An return process_notification(notification_type, priority=True, **kwargs) +# Retries for a transient backend problem (restart, 5xx). Kept inside the task +# because only the PG transport redelivers a failed message — on Celery a raise +# is terminal, so without this a rolling deploy would silently drop the email. +_GROUP_NOTIFICATION_ATTEMPTS = 3 +_GROUP_NOTIFICATION_RETRY_DELAY = 2.0 + + +def _post_group_notification(endpoint: str, organization_id: str, payload: dict) -> None: + """POST a group-notification job to the backend and insist it succeeded. + + Unlike ``_mark_buffer_outcome`` this deliberately **raises** on failure: + there is no reaper behind these rows, so a swallowed error would be a + silently unsent email. On the PG transport the raise also leaves the + message on the queue for redelivery, bounded by the consumer's attempt cap. + + A 4xx is not retried — a rejected payload will be rejected again. + """ + base_url = os.getenv("INTERNAL_API_BASE_URL") + api_key = os.getenv("INTERNAL_SERVICE_API_KEY") + if not base_url or not api_key: + raise RuntimeError( + "INTERNAL_API_BASE_URL / INTERNAL_SERVICE_API_KEY not set; " + "cannot send group notification" + ) + url = f"{base_url.rstrip('/')}/v1/group-notification/{endpoint}/" + headers = { + "Authorization": f"Bearer {api_key}", + # The backend resolves the tenant from this header; without it every + # org-scoped query comes back empty. + "X-Organization-ID": organization_id, + } + last_error = "" + for attempt in range(1, _GROUP_NOTIFICATION_ATTEMPTS + 1): + try: + with httpx.Client(transport=httpx.HTTPTransport(retries=2)) as client: + response = client.post(url, headers=headers, json=payload, timeout=30.0) + except Exception as e: # noqa: BLE001 - transport failure, retry below + last_error = f"exception={e!r}" + else: + if response.status_code == 200: + return + last_error = f"http_{response.status_code} body={response.text[:200]}" + if response.status_code < 500: + break + if attempt < _GROUP_NOTIFICATION_ATTEMPTS: + logger.warning( + "Group notification %s attempt %d/%d failed (%s); retrying", + endpoint, + attempt, + _GROUP_NOTIFICATION_ATTEMPTS, + last_error, + ) + time.sleep(_GROUP_NOTIFICATION_RETRY_DELAY) + raise RuntimeError(f"Group notification {endpoint} failed: {last_error}") + + +@worker_task(name="notify_resource_shared_with_group") +def notify_resource_shared_with_group( + group_ids: list[int], + actor_id: int, + resource_kind: str, + resource_id: str, + organization_id: str, +) -> None: + """Email every current member of the groups a resource was shared with.""" + _post_group_notification( + "resource-shared", + organization_id, + { + "group_ids": group_ids, + "actor_id": actor_id, + "resource_kind": resource_kind, + "resource_id": resource_id, + }, + ) + + +@worker_task(name="notify_group_membership_changed") +def notify_group_membership_changed( + group_id: int, + actor_id: int, + membership_action: str, + user_ids: list[int], + organization_id: str, +) -> None: + """Email the users whose membership of a group just changed.""" + _post_group_notification( + "membership-changed", + organization_id, + { + "group_id": group_id, + "actor_id": actor_id, + "membership_action": membership_action, + "user_ids": user_ids, + }, + ) + + @worker_task(name="notification_health_check") def notification_health_check() -> dict[str, Any]: """Health check task for notification worker.""" From d8b1008b70fa2010e28a772d0f2d71922cccfdc0 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Mon, 3 Aug 2026 19:26:01 +0530 Subject: [PATCH 02/10] UN-3494 [FIX] Restore direct-user sharing emails on the share endpoint UN-2977 moved sharing from PATCH to POST /{id}/share/, but the mixin's share action only diffed the groups axis. The per-viewset _notify_shared_users hooks stayed on partial_update, which nothing calls anymore, so sharing a resource with a user sent no email. Snapshot every declared axis and invoke the hook after the commit; declare it on the mixin as a no-op for hosts without a direct-share email. Co-Authored-By: Claude Opus 5 --- backend/permissions/resource_share_views.py | 29 ++++++++++++++------- 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/backend/permissions/resource_share_views.py b/backend/permissions/resource_share_views.py index f688569115..71030b0a7f 100644 --- a/backend/permissions/resource_share_views.py +++ b/backend/permissions/resource_share_views.py @@ -98,24 +98,36 @@ def share(self, request: Request, pk: str | None = None) -> Response: resource = self.get_object() # type: ignore[attr-defined] desired = _extract_desired_share_state(request.data) - # Only the groups axis is diffed: it is the one that notifies, and - # snapshotting ``shared_users`` too would fetch every viewer twice for - # nothing. Reads go through ``ResourceGroupShare``, so no refresh is - # needed between the two. - groups_before = self._read_axis(resource, "shared_groups") + before = self.snapshot_share_axes(resource) ShareAuthorizationService.authorize_and_commit( actor=request.user, resource=resource, desired=desired ) # ``_commit`` is the only atomic block on this path, so it has already - # committed — the diff reads persisted state and can never announce a + # committed — the diffs read persisted state and can never announce a # share that rolled back. notify_resource_shared_with_group( resource=resource, - groups=self._read_axis(resource, "shared_groups") - groups_before, + # ``.get`` — lookups narrows ``share_axes`` to users only. + groups=self._read_axis(resource, "shared_groups") + - before.get("shared_groups", set()), actor=request.user, ) + self._notify_shared_users(resource, before, request.data, request.user) return Response(status=status.HTTP_200_OK) + def _notify_shared_users( + self, + instance: Any, + before: dict[str, set[Any]], + request_data: dict[str, Any], + actor: Any, + /, + ) -> None: + """Email users newly added to ``shared_users``. + + Positional-only: hosts override with their own resource name and type. + """ + @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.""" @@ -132,8 +144,7 @@ def effective_members(self, request: Request, pk: str | None = None) -> Response 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. + Call BEFORE the write; pair with :meth:`diff_share_axes` afterward. """ return {axis: self._read_axis(instance, axis) for axis in self.share_axes} From 77ca124b5958f6ec34d5f9e68fe0af2bc1c78bee Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Tue, 4 Aug 2026 19:00:03 +0530 Subject: [PATCH 03/10] UN-3494 [FEAT] Email users and group members when resource access is revoked MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sharing already emailed on grant; revoking told nobody. Both axes now notify, and the seven duplicated copies of the user hook collapse into the share mixin. - ResourceShareManagementMixin gains a concrete _notify_shared_users covering grant and revoke, driven by the OwnerManagementMixin seam every host already declares. The seven per-viewset overrides and their dead partial_update wrappers go with it — a host override would otherwise shadow the mixin and silently swallow the revoke mail. - share() diffs both axes through _read_axis directly; AxisDiff, snapshot_share_axes, diff_share_axes and the share_axes ClassVar had no callers left. - Group revoke rides the existing resource-shared route with a share_action discriminator, mirroring membership-changed — no new endpoint or worker task. Defaulted at every hop so in-flight messages still run. - Suppressed when the user still reaches the resource via a group or shared_to_org: losing one axis is not losing access. Co-Authored-By: Claude Opus 5 --- backend/adapter_processor_v2/views.py | 47 ----- backend/api_v2/api_deployment_views.py | 37 ---- backend/connector_v2/views.py | 43 ----- backend/permissions/resource_share_views.py | 171 +++++++++++------- backend/pipeline_v2/views.py | 54 ------ .../prompt_studio_core_v2/views.py | 48 ----- .../group_notification_service.py | 11 +- backend/tenant_account_v2/internal_views.py | 8 +- .../tenant_account_v2/share_notifications.py | 45 ++++- backend/workflow_manager/workflow_v2/views.py | 45 ----- workers/notification/tasks.py | 7 +- 11 files changed, 161 insertions(+), 355 deletions(-) diff --git a/backend/adapter_processor_v2/views.py b/backend/adapter_processor_v2/views.py index f2aef82b0c..22f75d7b17 100644 --- a/backend/adapter_processor_v2/views.py +++ b/backend/adapter_processor_v2/views.py @@ -404,17 +404,6 @@ def destroy( raise DeleteAdapterInUseError(adapter_name=adapter_instance.adapter_name) return Response(status=status.HTTP_204_NO_CONTENT) - def partial_update( - self, request: Request, *args: tuple[Any], **kwargs: dict[str, Any] - ) -> Response: - adapter = self.get_object() - before = self.snapshot_share_axes(adapter) - - response = super().partial_update(request, *args, **kwargs) - if response.status_code == 200 and notification_plugin: - self._notify_shared_users(adapter, before, request.data, request.user) - return response - @action(detail=True, methods=["post"], url_path="share") def share(self, request: Request, pk: str | None = None) -> Response: """Apply share state, then clear default-adapter links for any user @@ -461,42 +450,6 @@ def on_owner_removed(self, resource: AdapterInstance, user: User) -> None: return self._clear_default_adapter_for_removed_users(resource, {user.pk}) - def _notify_shared_users( - self, - adapter: AdapterInstance, - before: dict[str, set[Any]], - request_data: dict[str, Any], - actor: Any, - ) -> None: - """Email users newly added to ``shared_users`` (best-effort).""" - users_diff = self.diff_share_axes(adapter, before, request_data).get( - "shared_users" - ) - if not (users_diff and users_diff.added): - return - try: - adapter_type_to_resource = { - "LLM": ResourceType.LLM.value, - "EMBEDDING": ResourceType.EMBEDDING.value, - "VECTOR_DB": ResourceType.VECTOR_DB.value, - "X2TEXT": ResourceType.X2TEXT.value, - } - resource_type = adapter_type_to_resource.get( - adapter.adapter_type, ResourceType.LLM.value - ) - service_class = notification_plugin["service_class"] - notification_service = service_class() - notification_service.send_sharing_notification( - resource_type=resource_type, - resource_name=adapter.adapter_name, - resource_id=str(adapter.id), - shared_by=actor, - shared_to=list(users_diff.added), - resource_instance=adapter, - ) - except Exception as e: - logger.exception("Failed to send sharing notification: %s", e) - def _clear_default_adapter_for_removed_users( self, adapter: AdapterInstance, diff --git a/backend/api_v2/api_deployment_views.py b/backend/api_v2/api_deployment_views.py index 8b96bcd67e..f9106e312f 100644 --- a/backend/api_v2/api_deployment_views.py +++ b/backend/api_v2/api_deployment_views.py @@ -414,40 +414,3 @@ def list_of_shared_users(self, request: Request, pk: str | None = None) -> Respo instance = self.get_object() serializer = SharedUserListSerializer(instance) return Response(serializer.data) - - def partial_update(self, request: Request, *args: Any, **kwargs: Any) -> Response: - """Override partial_update to handle sharing notifications.""" - instance = self.get_object() - before = self.snapshot_share_axes(instance) - - response = super().partial_update(request, *args, **kwargs) - if response.status_code == 200 and notification_plugin: - self._notify_shared_users(instance, before, request.data, request.user) - return response - - def _notify_shared_users( - self, - instance: APIDeployment, - before: dict[str, set[Any]], - request_data: dict[str, Any], - actor: Any, - ) -> None: - """Email users newly added to ``shared_users`` (best-effort).""" - users_diff = self.diff_share_axes(instance, before, request_data).get( - "shared_users" - ) - if not (users_diff and users_diff.added): - return - try: - service_class = notification_plugin["service_class"] - notification_service = service_class() - notification_service.send_sharing_notification( - resource_type=ResourceType.API_DEPLOYMENT.value, - resource_name=instance.display_name, - resource_id=str(instance.id), - shared_by=actor, - shared_to=list(users_diff.added), - resource_instance=instance, - ) - except Exception as e: - logger.exception("Failed to send sharing notification: %s", e) diff --git a/backend/connector_v2/views.py b/backend/connector_v2/views.py index fd75b749db..c3bb018ff1 100644 --- a/backend/connector_v2/views.py +++ b/backend/connector_v2/views.py @@ -36,7 +36,6 @@ notification_plugin = get_plugin("notification") if notification_plugin: from plugins.notification.constants import ResourceType - from plugins.notification.sharing_notification import SharingNotificationService logger = logging.getLogger(__name__) @@ -286,45 +285,3 @@ def perform_destroy(self, instance: ConnectorInstance) -> None: f" named {instance.connector_name}" ) raise DeleteConnectorInUseError(connector_name=instance.connector_name) - - def partial_update(self, request: Request, *args: Any, **kwargs: Any) -> Response: - """Override to handle sharing notifications.""" - instance = self.get_object() - before = self.snapshot_share_axes(instance) - - response = super().partial_update(request, *args, **kwargs) - if response.status_code == 200 and notification_plugin: - self._notify_shared_users(instance, before, request.data, request.user) - return response - - def _notify_shared_users( - self, - instance: ConnectorInstance, - before: dict[str, set[Any]], - request_data: dict[str, Any], - actor: Any, - ) -> None: - """Email users newly added to ``shared_users`` (best-effort).""" - users_diff = self.diff_share_axes(instance, before, request_data).get( - "shared_users" - ) - if not (users_diff and users_diff.added): - return - try: - SharingNotificationService().send_sharing_notification( - resource_type=ResourceType.CONNECTOR.value, - resource_name=instance.connector_name, - resource_id=str(instance.id), - shared_by=actor, - shared_to=list(users_diff.added), - resource_instance=instance, - ) - logger.info( - "Sent sharing notifications for connector to %d users", - len(users_diff.added), - ) - except Exception as e: - logger.exception( - "Failed to send sharing notification, continuing update though: %s", - str(e), - ) diff --git a/backend/permissions/resource_share_views.py b/backend/permissions/resource_share_views.py index 71030b0a7f..3b13cdc988 100644 --- a/backend/permissions/resource_share_views.py +++ b/backend/permissions/resource_share_views.py @@ -1,22 +1,26 @@ """Shared share-management surface for resource ViewSets. -The mixin is **axis-agnostic** — it operates over the sharing "axes" declared -in :attr:`ResourceShareManagementMixin.share_axes`. ``shared_users`` is an M2M -on the resource model, while ``shared_groups`` is stored polymorphically in -``ResourceGroupShare`` (not an M2M) and routed through the sharing helpers; new -axes can be added by extending that attribute. +The mixin is **axis-agnostic** — it reads the sharing "axes" named in +``_SUPPORTED_SHARE_AXES``. ``shared_users`` is the direct-viewer axis, backed by +VIEWER membership rows, while ``shared_groups`` is stored polymorphically in +``ResourceGroupShare`` (not an M2M) and routed through the sharing helpers. """ -from dataclasses import dataclass, field -from typing import Any, ClassVar +import logging +from typing import Any from django.db.models import Model +from plugins import get_plugin from rest_framework import status from rest_framework.decorators import action from rest_framework.exceptions import ValidationError from rest_framework.request import Request from rest_framework.response import Response +logger = logging.getLogger(__name__) + +notification_plugin = get_plugin("notification") + _SUPPORTED_SHARE_AXES = ("shared_users", "shared_groups", "shared_to_org") @@ -55,30 +59,78 @@ def _coerce_id_list(axis: str, value: Any) -> list[int]: return coerced -@dataclass -class AxisDiff: - """Pre/post snapshot for a single share axis (M2M field).""" - - before: set[Any] = field(default_factory=set) - after: set[Any] = field(default_factory=set) +def _notification_context(view: Any, instance: Any) -> tuple[str, str] | None: + """Resolve ``(resource_type, resource_name)`` for the email senders. - @property - def added(self) -> set[Any]: - return self.after - self.before - - @property - def removed(self) -> set[Any]: - return self.before - self.after + ``None`` when the plugin is absent or the host ViewSet has not opted in by + setting ``notification_resource_name_field`` and overriding + ``get_notification_resource_type`` (both declared on + ``OwnerManagementMixin``, which every share host also mixes in). + """ + name_field = getattr(view, "notification_resource_name_field", None) + resolve_type = getattr(view, "get_notification_resource_type", None) + if not notification_plugin or not name_field or resolve_type is None: + return None + resource_type = resolve_type(instance) + resource_name = getattr(instance, name_field, None) + if resource_type is None or not resource_name: + return None + return resource_type, resource_name + + +def _users_left_without_access(instance: Model, users: set[Any]) -> list[Any]: + """Narrow ``users`` to those with no remaining access to ``instance``. + + Someone dropped from ``shared_users`` may still reach the resource via a + group or an org-wide share; telling them their access was removed would be + wrong. + """ + if not users: + return [] + from tenant_account_v2.sharing_helpers import compute_effective_members + + retained = {member["user_id"] for member in compute_effective_members(instance)} + return [user for user in users if user.pk not in retained] + + +def _send_share_notification( + instance: Model, context: tuple[str, str], users: set[Any], actor: Any +) -> None: + """Email users newly granted direct access. Best-effort.""" + resource_type, resource_name = context + try: + notification_plugin["service_class"]().send_sharing_notification( + resource_type=resource_type, + resource_name=resource_name, + resource_id=str(instance.pk), + shared_by=actor, + shared_to=list(users), + resource_instance=instance, + ) + except Exception: + logger.exception("Failed to send sharing notification for %s", instance.pk) + + +def _send_revoke_notification( + instance: Model, context: tuple[str, str], users: list[Any], actor: Any +) -> None: + """Email users whose direct access was revoked. Best-effort.""" + resource_type, resource_name = context + try: + notification_plugin["service_class"]().send_access_removed_notification( + resource_type=resource_type, + resource_name=resource_name, + resource_id=str(instance.pk), + removed_from=users, + removed_by=actor, + resource_instance=instance, + ) + except Exception: + logger.exception("Failed to send access-removed notification for %s", instance.pk) class ResourceShareManagementMixin: - """Adds the shared share-management surface to a resource ViewSet. - - Subclasses declare share axes via :attr:`share_axes`. The default - covers ``shared_users`` + ``shared_groups``. - """ - - share_axes: ClassVar[tuple[str, ...]] = ("shared_users", "shared_groups") + """Adds the shared share-management surface to a resource ViewSet.""" @action(detail=True, methods=["post"], url_path="share") def share(self, request: Request, pk: str | None = None) -> Response: @@ -92,41 +144,55 @@ def share(self, request: Request, pk: str | None = None) -> Response: ``ShareAuthorizationService``. """ from tenant_account_v2.share_notifications import ( - notify_resource_shared_with_group, + 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) - before = self.snapshot_share_axes(resource) + users_before = self._read_axis(resource, "shared_users") + groups_before = self._read_axis(resource, "shared_groups") ShareAuthorizationService.authorize_and_commit( actor=request.user, resource=resource, desired=desired ) # ``_commit`` is the only atomic block on this path, so it has already # committed — the diffs read persisted state and can never announce a # share that rolled back. - notify_resource_shared_with_group( + resource.refresh_from_db() + users_after = self._read_axis(resource, "shared_users") + groups_after = self._read_axis(resource, "shared_groups") + notify_resource_group_share_changed( resource=resource, - # ``.get`` — lookups narrows ``share_axes`` to users only. - groups=self._read_axis(resource, "shared_groups") - - before.get("shared_groups", set()), + added=groups_after - groups_before, + removed=groups_before - groups_after, actor=request.user, ) - self._notify_shared_users(resource, before, request.data, 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, - before: dict[str, set[Any]], - request_data: dict[str, Any], + added: set[Any], + removed: set[Any], actor: Any, /, ) -> None: - """Email users newly added to ``shared_users``. + """Email users granted or denied direct access. - Positional-only: hosts override with their own resource name and type. + Resource type and name come from the host's ``OwnerManagementMixin`` + seam, so every share host is covered without an override. """ + context = _notification_context(self, instance) + if context is None: + return + if added: + _send_share_notification(instance, context, added, actor) + revoked = _users_left_without_access(instance, removed) + if revoked: + _send_revoke_notification(instance, context, revoked, actor) @action(detail=True, methods=["get"], url_path="effective-members") def effective_members(self, request: Request, pk: str | None = None) -> Response: @@ -141,35 +207,6 @@ def effective_members(self, request: Request, pk: str | None = None) -> Response members = compute_effective_members(self.get_object()) # type: ignore[attr-defined] return Response(EffectiveMemberSerializer(members, many=True).data) - def snapshot_share_axes(self, instance: Model) -> dict[str, set[Any]]: - """Capture every declared axis's current contents. - - Call BEFORE the write; 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/pipeline_v2/views.py b/backend/pipeline_v2/views.py index ec7e7720f3..2683ac2bcb 100644 --- a/backend/pipeline_v2/views.py +++ b/backend/pipeline_v2/views.py @@ -187,60 +187,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 cd65f2de77..6328293b0c 100644 --- a/backend/prompt_studio/prompt_studio_core_v2/views.py +++ b/backend/prompt_studio/prompt_studio_core_v2/views.py @@ -335,54 +335,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 index 53f2bc7f82..ccdc3d9989 100644 --- a/backend/tenant_account_v2/group_notification_service.py +++ b/backend/tenant_account_v2/group_notification_service.py @@ -20,7 +20,7 @@ from plugins import get_plugin from tenant_account_v2.models import OrganizationGroup, OrganizationMember -from tenant_account_v2.share_notifications import MembershipAction +from tenant_account_v2.share_notifications import MembershipAction, ShareAction from tenant_account_v2.shareable_resources import ShareableResource, descriptor_for_kind if TYPE_CHECKING: @@ -62,11 +62,12 @@ def send_resource_shared( actor_id: int, resource_kind: str, resource_id: str, + share_action: str = ShareAction.SHARED.value, ) -> None: - """Mail every current member of each group that a resource was shared. + """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. + the recipient actually belongs to. ``share_action`` picks the wording. """ service = _service() if service is None: @@ -90,9 +91,10 @@ def send_resource_shared( organization, group.memberships.values_list("user_id", flat=True) ) logger.info( - "group-notification: task=%s group_id=%s recipient_count=%d", + "group-notification: task=%s group_id=%s action=%s recipient_count=%d", "notify_resource_shared_with_group", group.pk, + share_action, len(recipients), ) if not recipients: @@ -105,6 +107,7 @@ def send_resource_shared( shared_by=actor, shared_to=recipients, resource_instance=resource, + share_action=ShareAction(share_action).value, ) diff --git a/backend/tenant_account_v2/internal_views.py b/backend/tenant_account_v2/internal_views.py index fb707df14e..0e959c9b5d 100644 --- a/backend/tenant_account_v2/internal_views.py +++ b/backend/tenant_account_v2/internal_views.py @@ -25,7 +25,7 @@ send_membership_changed, send_resource_shared, ) -from tenant_account_v2.share_notifications import MembershipAction +from tenant_account_v2.share_notifications import MembershipAction, ShareAction logger = logging.getLogger(__name__) @@ -37,6 +37,10 @@ class ResourceSharedWithGroupSerializer(serializers.Serializer): actor_id = serializers.IntegerField() resource_kind = serializers.CharField() resource_id = serializers.CharField() + # Defaulted so messages enqueued before this field existed still validate. + share_action = serializers.ChoiceField( + choices=[a.value for a in ShareAction], default=ShareAction.SHARED.value + ) class GroupMembershipChangedSerializer(serializers.Serializer): @@ -64,7 +68,7 @@ def _organization() -> Organization: class ResourceSharedWithGroupView(_GroupNotificationView): - """Mail every current member of the groups a resource was just shared with.""" + """Mail every current member of the groups whose resource access just changed.""" def post(self, request: Request) -> Response: serializer = ResourceSharedWithGroupSerializer(data=request.data) diff --git a/backend/tenant_account_v2/share_notifications.py b/backend/tenant_account_v2/share_notifications.py index c4191a8a80..526f926a04 100644 --- a/backend/tenant_account_v2/share_notifications.py +++ b/backend/tenant_account_v2/share_notifications.py @@ -1,8 +1,8 @@ """Enqueue hooks for group-sharing email notifications (UN-3494 / mfbt UNS-848). -Two events earn a group's members an email: a resource shared with 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. +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 @@ -53,14 +53,44 @@ class MembershipAction(StrEnum): REMOVED = "removed" -def notify_resource_shared_with_group( - *, resource: Any, groups: Iterable[OrganizationGroup], actor: User +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 "a resource was shared with your group" mail for newly added groups. + """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. + 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. """ group_ids = sorted(group.pk for group in groups) if not group_ids: @@ -76,6 +106,7 @@ def notify_resource_shared_with_group( "actor_id": actor.pk, "resource_kind": kind, "resource_id": str(resource.pk), + "share_action": str(share_action), "organization_id": organization_id, }, organization_id=organization_id, diff --git a/backend/workflow_manager/workflow_v2/views.py b/backend/workflow_manager/workflow_v2/views.py index fefba8c21a..e567f39e2b 100644 --- a/backend/workflow_manager/workflow_v2/views.py +++ b/backend/workflow_manager/workflow_v2/views.py @@ -173,51 +173,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/workers/notification/tasks.py b/workers/notification/tasks.py index f6e3d1ee9c..10945e34c8 100644 --- a/workers/notification/tasks.py +++ b/workers/notification/tasks.py @@ -531,8 +531,12 @@ def notify_resource_shared_with_group( resource_kind: str, resource_id: str, organization_id: str, + share_action: str = "shared", ) -> None: - """Email every current member of the groups a resource was shared with.""" + """Email every current member of the groups whose access just changed. + + ``share_action`` defaults so messages enqueued before it existed still run. + """ _post_group_notification( "resource-shared", organization_id, @@ -541,6 +545,7 @@ def notify_resource_shared_with_group( "actor_id": actor_id, "resource_kind": resource_kind, "resource_id": resource_id, + "share_action": share_action, }, ) From 3392ebea55fa9cb21c5a026d12c27f5a86cfd239 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Wed, 5 Aug 2026 13:41:22 +0530 Subject: [PATCH 04/10] UN-3494 [FIX] Gate co-owner removal behind the share modal's Apply button MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adding a co-owner was staged until Apply, but revoking one fired the DELETE straight from the Popconfirm — so Cancel could not undo it, Apply stayed disabled for a removal-only edit, and the revoke email went out on click. Stage the roster the way SharePermission does: one selected-owners list seeded from the server, edited locally by both add and revoke, committed only by Apply. Collapse the hook's two mutation callbacks into one onApplyCoOwners that runs adds before removes (so a one-shot owner swap clears the backend's last-owner guard), refreshes once, and emits one summary alert. Apply now closes on a clean run and stays open on failure, matching useShareModal. Co-Authored-By: Claude Opus 5 --- .../api-deployment/ApiDeployment.jsx | 7 +- .../pipelines/Pipelines.jsx | 7 +- .../co-owner-management/CoOwnerManagement.css | 4 - .../co-owner-management/CoOwnerManagement.jsx | 218 ++++++++---------- .../co-owner-management/CoOwnerModal.jsx | 4 +- frontend/src/hooks/useCoOwnerManagement.jsx | 139 ++++++----- 6 files changed, 171 insertions(+), 208 deletions(-) diff --git a/frontend/src/components/deployments/api-deployment/ApiDeployment.jsx b/frontend/src/components/deployments/api-deployment/ApiDeployment.jsx index b1e5bd0708..4f2b092975 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, @@ -408,10 +407,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/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 7d8648e13a..bdc58ed206 100644 --- a/frontend/src/components/widgets/co-owner-management/CoOwnerManagement.jsx +++ b/frontend/src/components/widgets/co-owner-management/CoOwnerManagement.jsx @@ -13,7 +13,7 @@ import { Typography, } from "antd"; import PropTypes from "prop-types"; -import { useMemo, useState } from "react"; +import { useEffect, useMemo, useState } from "react"; import { SpinnerLoader } from "../spinner-loader/SpinnerLoader"; import "./CoOwnerManagement.css"; @@ -26,82 +26,84 @@ 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. + useEffect(() => { + setSelectedOwners(ownersList); + }, [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 so + // the user can see what was rejected and retry. + 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 ( - {loading || applying ? ( + {loading ? ( ) : ( <> @@ -134,77 +136,56 @@ function CoOwnerManagement({ }))} /> 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 && ( +
event.stopPropagation()} + role="none" + > + } + onConfirm={() => handleRemove(item?.id)} + > +
+ ) + } + > + + } /> - ) : ( - totalOwners > 1 && ( -
event.stopPropagation()} - role="none" - > - } - onConfirm={() => handleRemoveExisting(item?.id)} - > -
- ) - ) + + {item.email} + + } - > - - } - /> - - {item.email} - - - } - /> - - ); - }} + /> + + )} /> ) : ( No co-owners yet @@ -218,13 +199,12 @@ function CoOwnerManagement({ CoOwnerManagement.propTypes = { open: PropTypes.bool.isRequired, setOpen: PropTypes.func.isRequired, - resourceId: PropTypes.string.isRequired, + resourceId: PropTypes.string, resourceType: PropTypes.string.isRequired, allUsers: PropTypes.array, coOwners: PropTypes.array, loading: PropTypes.bool, - onAddCoOwner: PropTypes.func.isRequired, - onRemoveCoOwner: PropTypes.func.isRequired, + onApplyCoOwners: PropTypes.func.isRequired, }; export { CoOwnerManagement }; diff --git a/frontend/src/components/widgets/co-owner-management/CoOwnerModal.jsx b/frontend/src/components/widgets/co-owner-management/CoOwnerModal.jsx index e66be15223..ccadfd0b82 100644 --- a/frontend/src/components/widgets/co-owner-management/CoOwnerModal.jsx +++ b/frontend/src/components/widgets/co-owner-management/CoOwnerModal.jsx @@ -21,10 +21,8 @@ function CoOwnerModal({ coOwner, resourceType }) { resourceType={resourceType} allUsers={coOwner.coOwnerAllUsers} coOwners={coOwner.coOwnerData.coOwners} - createdBy={coOwner.coOwnerData.createdBy} loading={coOwner.coOwnerLoading} - onAddCoOwner={coOwner.onAddCoOwner} - onRemoveCoOwner={coOwner.onRemoveCoOwner} + onApplyCoOwners={coOwner.onApplyCoOwners} /> ); } diff --git a/frontend/src/hooks/useCoOwnerManagement.jsx b/frontend/src/hooks/useCoOwnerManagement.jsx index 5a3b7e5ef1..c3fee93d10 100644 --- a/frontend/src/hooks/useCoOwnerManagement.jsx +++ b/frontend/src/hooks/useCoOwnerManagement.jsx @@ -2,14 +2,47 @@ import { useCallback, useRef, useState } from "react"; import { useExceptionHandler } from "./useExceptionHandler"; +/** + * Summarize one Apply into a single alert. + * + * Failures carry the user object rather than the id, so an owner who has since + * left the org — and is therefore missing from the org member list — is still + * named by email. + */ +function buildApplyAlert( + addUsers, + removeUsers, + failed, + lastError, + handleException, +) { + const total = addUsers.length + removeUsers.length; + if (failed.length === total) { + return handleException(lastError, "Unable to update co-owners"); + } + const failedIds = new Set(failed.map((user) => String(user?.id))); + const done = (users) => + users.filter((user) => !failedIds.has(String(user?.id))).length; + const parts = []; + if (done(addUsers)) { + parts.push(`${done(addUsers)} added`); + } + if (done(removeUsers)) { + parts.push(`${done(removeUsers)} removed`); + } + const summary = `Co-owners updated: ${parts.join(", ")}`; + if (failed.length === 0) { + return { type: "success", content: summary }; + } + const failedNames = failed.map((user) => user?.email || user?.id).join(", "); + return { type: "warning", content: `${summary}. Failed for: ${failedNames}` }; +} + function useCoOwnerManagement({ service, setAlertDetails, onListRefresh }) { const handleException = useExceptionHandler(); const [coOwnerOpen, setCoOwnerOpen] = useState(false); - const [coOwnerData, setCoOwnerData] = useState({ - coOwners: [], - createdBy: null, - }); + const [coOwnerData, setCoOwnerData] = useState({ coOwners: [] }); const [coOwnerLoading, setCoOwnerLoading] = useState(false); const [coOwnerAllUsers, setCoOwnerAllUsers] = useState([]); const [coOwnerResourceId, setCoOwnerResourceId] = useState(null); @@ -25,10 +58,7 @@ function useCoOwnerManagement({ service, setAlertDetails, onListRefresh }) { try { const res = await service.getSharedUsers(resourceId); if (latestRequestRef.current !== requestId) return; - setCoOwnerData({ - coOwners: res.data?.co_owners || [], - createdBy: res.data?.created_by || null, - }); + setCoOwnerData({ coOwners: res.data?.co_owners || [] }); } catch (err) { if (latestRequestRef.current !== requestId) return; if (err?.response?.status === 404) { @@ -74,7 +104,6 @@ function useCoOwnerManagement({ service, setAlertDetails, onListRefresh }) { setCoOwnerAllUsers(userList); setCoOwnerData({ coOwners: sharedUsersResponse.data?.co_owners || [], - createdBy: sharedUsersResponse.data?.created_by || null, }); } catch (err) { if (latestRequestRef.current !== requestId) return; @@ -91,73 +120,40 @@ function useCoOwnerManagement({ service, setAlertDetails, onListRefresh }) { [service, setAlertDetails, handleException], ); - const onAddCoOwner = useCallback( - async (resourceId, userIdOrIds) => { + const onApplyCoOwners = useCallback( + async (resourceId, { addUsers = [], removeUsers = [] }) => { const requestId = latestRequestRef.current; - const isBatch = Array.isArray(userIdOrIds); - const userIds = isBatch ? userIdOrIds : [userIdOrIds]; - // Attempt every id independently — a mid-batch failure must not drop the - // remaining ids or contradict the refreshed modal state. - const failedIds = []; + // Attempt every user independently — one rejection must not drop the rest + // or leave the modal contradicting the server. + const failed = []; let lastError = null; - for (const userId of userIds) { - try { - await service.addCoOwner(resourceId, userId); - } catch (err) { - failedIds.push(userId); - lastError = err; + const run = async (users, call) => { + for (const user of users) { + try { + await call(user.id); + } catch (err) { + failed.push(user); + lastError = err; + } } - } + }; + // Adds first: the backend rejects removing the last owner, so a one-shot + // owner swap has to grow the roster before it shrinks it. + await run(addUsers, (id) => service.addCoOwner(resourceId, id)); + await run(removeUsers, (id) => service.removeCoOwner(resourceId, id)); // Reconverge the modal on true server state regardless of partial outcome. await refreshCoOwnerData(resourceId, requestId); onListRefresh?.(); - - const succeeded = userIds.length - failedIds.length; - if (failedIds.length === 0) { - setAlertDetails({ - type: "success", - content: isBatch - ? "Co-owners added successfully" - : "Co-owner added successfully", - }); - } else if (succeeded === 0) { - setAlertDetails(handleException(lastError, "Unable to add co-owner")); - } else { - const failedEmails = coOwnerAllUsers - .filter((user) => failedIds.includes(user.id)) - .map((user) => user.email); - setAlertDetails({ - type: "warning", - content: `Added ${succeeded} of ${userIds.length} co-owners. Failed for: ${ - failedEmails.join(", ") || failedIds.join(", ") - }`, - }); - } - }, - [ - service, - refreshCoOwnerData, - onListRefresh, - setAlertDetails, - handleException, - coOwnerAllUsers, - ], - ); - - const onRemoveCoOwner = useCallback( - async (resourceId, userId) => { - const requestId = latestRequestRef.current; - try { - await service.removeCoOwner(resourceId, userId); - setAlertDetails({ - type: "success", - content: "Co-owner removed successfully", - }); - await refreshCoOwnerData(resourceId, requestId); - onListRefresh?.(); - } catch (err) { - setAlertDetails(handleException(err, "Unable to remove co-owner")); - } + setAlertDetails( + buildApplyAlert( + addUsers, + removeUsers, + failed, + lastError, + handleException, + ), + ); + return failed.length === 0; }, [ service, @@ -176,8 +172,7 @@ function useCoOwnerManagement({ service, setAlertDetails, onListRefresh }) { coOwnerAllUsers, coOwnerResourceId, handleCoOwner, - onAddCoOwner, - onRemoveCoOwner, + onApplyCoOwners, }; } From c1d30955e58e3c9adb170e2e605c71ca5e432994 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Wed, 5 Aug 2026 18:39:26 +0530 Subject: [PATCH 05/10] UN-3494 [FIX] Address review findings on sharing notifications MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Group revoke no longer mails members who kept access another way. The revoke recipient list now runs through the same effective-access filter the direct path uses, with owners folded in — compute_effective_members excludes them by design, and the sharer is usually a member of the group they shared with, so revoking told the owner their own access was removed and pointed them at the dashboard. - _get_user is org-scoped through OrganizationMember, the one unscoped query left on this tenant path. Service accounts are kept so a platform-account share still notifies. - _notify_shared_users is wrapped: the share has already committed by the time it runs, so a raising seam or a DB hiccup must not 500 a share that worked. - _users_left_without_access short-circuits on shared_to_org — nobody lost access, and answering it otherwise hydrates every member of the org. - _notification_context loses its duplicate copy and uses the OwnerManagementMixin definition every host already inherits. - Logs the Flipt decision, and how many recipients were dropped versus requested, so a missing email is diagnosable. - Docstrings corrected: the mixin is not axis-agnostic, transport resolves per resource id not per org, the flag is evaluated once at enqueue, and delivery is at-least-once. - Sonar S7632: the noqa directive carried trailing prose. - Co-owner apply no longer overwrites the resource-gone alert with its summary. Co-Authored-By: Claude Opus 5 --- backend/permissions/membership_views.py | 5 + backend/permissions/resource_share_views.py | 57 ++++---- .../group_notification_service.py | 127 +++++++++++++----- backend/tenant_account_v2/internal_views.py | 2 +- .../tenant_account_v2/share_notifications.py | 27 ++-- frontend/src/hooks/useCoOwnerManagement.jsx | 11 +- workers/notification/tasks.py | 7 +- 7 files changed, 159 insertions(+), 77 deletions(-) diff --git a/backend/permissions/membership_views.py b/backend/permissions/membership_views.py index e9ade97e9c..329c6a3aa5 100644 --- a/backend/permissions/membership_views.py +++ b/backend/permissions/membership_views.py @@ -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) diff --git a/backend/permissions/resource_share_views.py b/backend/permissions/resource_share_views.py index 3b13cdc988..fbc61de1b7 100644 --- a/backend/permissions/resource_share_views.py +++ b/backend/permissions/resource_share_views.py @@ -1,9 +1,9 @@ """Shared share-management surface for resource ViewSets. -The mixin is **axis-agnostic** — it reads the sharing "axes" named in -``_SUPPORTED_SHARE_AXES``. ``shared_users`` is the direct-viewer axis, backed by -VIEWER membership rows, while ``shared_groups`` is stored polymorphically in -``ResourceGroupShare`` (not an M2M) and routed through the sharing helpers. +The mixin reads the sharing "axes" named in ``_SUPPORTED_SHARE_AXES``. +``shared_users`` is the direct-viewer axis, backed by VIEWER membership rows, +while ``shared_groups`` is stored polymorphically in ``ResourceGroupShare`` +(not an M2M) and routed through the sharing helpers. """ import logging @@ -59,25 +59,6 @@ def _coerce_id_list(axis: str, value: Any) -> list[int]: return coerced -def _notification_context(view: Any, instance: Any) -> tuple[str, str] | None: - """Resolve ``(resource_type, resource_name)`` for the email senders. - - ``None`` when the plugin is absent or the host ViewSet has not opted in by - setting ``notification_resource_name_field`` and overriding - ``get_notification_resource_type`` (both declared on - ``OwnerManagementMixin``, which every share host also mixes in). - """ - name_field = getattr(view, "notification_resource_name_field", None) - resolve_type = getattr(view, "get_notification_resource_type", None) - if not notification_plugin or not name_field or resolve_type is None: - return None - resource_type = resolve_type(instance) - resource_name = getattr(instance, name_field, None) - if resource_type is None or not resource_name: - return None - return resource_type, resource_name - - def _users_left_without_access(instance: Model, users: set[Any]) -> list[Any]: """Narrow ``users`` to those with no remaining access to ``instance``. @@ -87,6 +68,11 @@ def _users_left_without_access(instance: Model, users: set[Any]) -> list[Any]: """ if not users: return [] + if getattr(instance, "shared_to_org", False): + # Org-wide share still covers everyone — nobody lost access, and + # answering it via ``compute_effective_members`` would hydrate every + # member of the org to say so. + return [] from tenant_account_v2.sharing_helpers import compute_effective_members retained = {member["user_id"] for member in compute_effective_members(instance)} @@ -180,19 +166,24 @@ def _notify_shared_users( actor: Any, /, ) -> None: - """Email users granted or denied direct access. + """Email users granted or denied direct access. Best-effort. Resource type and name come from the host's ``OwnerManagementMixin`` - seam, so every share host is covered without an override. + 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. """ - context = _notification_context(self, instance) - if context is None: - return - if added: - _send_share_notification(instance, context, added, actor) - revoked = _users_left_without_access(instance, removed) - if revoked: - _send_revoke_notification(instance, context, revoked, actor) + 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: diff --git a/backend/tenant_account_v2/group_notification_service.py b/backend/tenant_account_v2/group_notification_service.py index ccdc3d9989..7030d642f2 100644 --- a/backend/tenant_account_v2/group_notification_service.py +++ b/backend/tenant_account_v2/group_notification_service.py @@ -12,6 +12,7 @@ from __future__ import annotations import logging +from dataclasses import dataclass from typing import TYPE_CHECKING, Any from account_v2.models import Organization, User @@ -55,6 +56,15 @@ 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, @@ -72,43 +82,30 @@ def send_resource_shared( service = _service() if service is None: return - actor = _get_user(actor_id) - resource, resource_name, resource_type = _load_resource( - organization, resource_kind, resource_id - ) - if actor is None or resource_type is None: + actor = _get_user(organization, actor_id) + shared = _load_resource(organization, resource_kind, resource_id) + if actor is None or shared.type is None: logger.info( "group-notification: skipping resource share for %s/%s " "(actor_found=%s resource_type=%s)", resource_kind, resource_id, actor is not None, - resource_type, + shared.type, ) return + retained = _retained_user_ids(shared.instance, share_action) for group in _groups_in_org(organization, group_ids): - recipients = _live_member_users( - organization, group.memberships.values_list("user_id", flat=True) - ) + recipients = _group_recipients(organization, group, retained) logger.info( - "group-notification: task=%s group_id=%s action=%s recipient_count=%d", - "notify_resource_shared_with_group", + "group-notification: task=notify_resource_shared_with_group " + "group_id=%s action=%s recipient_count=%d", group.pk, share_action, len(recipients), ) - if not recipients: - continue - service.send_group_resource_shared_notification( - resource_type=resource_type, - resource_name=resource_name, - resource_id=str(resource.pk), - group_name=group.name, - shared_by=actor, - shared_to=recipients, - resource_instance=resource, - share_action=ShareAction(share_action).value, - ) + if recipients: + _mail_group(service, group, recipients, shared, actor, share_action) def send_membership_changed( @@ -129,7 +126,7 @@ def send_membership_changed( service = _service() if service is None: return - actor = _get_user(actor_id) + actor = _get_user(organization, actor_id) group = _groups_in_org(organization, [group_id]).first() if actor is None or group is None: logger.info( @@ -167,8 +164,67 @@ def _service() -> Any | None: return notification_plugin["service_class"]() -def _get_user(user_id: int) -> User | None: - return User.objects.filter(pk=user_id).first() +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(resource: Any, share_action: str) -> set[int]: + """Users who still reach ``resource``; empty on the share direction. + + A revoked group's members may keep access through another group, a direct + share or an org-wide share — telling them it was removed would be wrong, + and the revoke email also repoints their CTA at the dashboard. Owners sit + outside ``compute_effective_members`` by design, so add them back: an owner + in the revoked group has lost nothing. + """ + if share_action != ShareAction.REVOKED.value: + return set() + from tenant_account_v2.sharing_helpers import compute_effective_members + + return {member["user_id"] for member in compute_effective_members(resource)} | { + owner.pk for owner in resource.owners() + } + + +def _group_recipients( + organization: Organization, group: OrganizationGroup, retained: set[int] +) -> list[User]: + """Live members of ``group`` who did not keep access via ``retained``.""" + users = _live_member_users( + organization, group.memberships.values_list("user_id", flat=True) + ) + return [user for user in users if user.pk not in retained] + + +def _mail_group( + service: Any, + group: OrganizationGroup, + recipients: list[User], + shared: _SharedResource, + actor: User, + share_action: str, +) -> None: + """Send one group's copy of the resource-share email.""" + 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( @@ -185,20 +241,29 @@ def _live_member_users(organization: Organization, user_ids: Iterable[int]) -> l Service accounts are excluded, matching ``compute_effective_members``. """ + requested = list(user_ids) memberships = OrganizationMember.objects.filter( - organization=organization, user_id__in=list(user_ids) + organization=organization, user_id__in=requested ).select_related("user") - return [ + 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 -) -> tuple[Any, str, str | None]: - """Resolve the shared resource to ``(instance, display name, plugin type)``. +) -> _SharedResource: + """Resolve the shared resource for the email senders. Raises: ResourceNotFoundError: the descriptor, model, or row is missing — the @@ -220,7 +285,7 @@ def _load_resource( if resource is None: raise ResourceNotFoundError(f"{kind} {resource_id} not found in organization") name = getattr(resource, descriptor.name_field, "") or "" - return resource, name, _resource_type_for(descriptor, resource) + return _SharedResource(resource, name, _resource_type_for(descriptor, resource)) def _resource_type_for(descriptor: ShareableResource, resource: Any) -> str | None: diff --git a/backend/tenant_account_v2/internal_views.py b/backend/tenant_account_v2/internal_views.py index 0e959c9b5d..c5797b53c0 100644 --- a/backend/tenant_account_v2/internal_views.py +++ b/backend/tenant_account_v2/internal_views.py @@ -6,7 +6,7 @@ email plugin — needs it. Failure contract: **any** unhandled problem must surface as non-2xx so the -queue redelivers. The one deliberate exception is a resource that no longer +worker retries. The one deliberate exception is a resource that no longer exists, which returns 200 — retrying that can only fail again. """ diff --git a/backend/tenant_account_v2/share_notifications.py b/backend/tenant_account_v2/share_notifications.py index 526f926a04..50155b0aba 100644 --- a/backend/tenant_account_v2/share_notifications.py +++ b/backend/tenant_account_v2/share_notifications.py @@ -6,13 +6,16 @@ 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 resolved per-org by the same -``resolve_transport`` gate the execution path uses — the PG queue where that is -enabled, Celery otherwise. - -The whole feature sits behind its own Flipt flag and fails closed everywhere: a -blind Flipt, a missing org, or any dispatch error means no notification, never -a broken share. +backend does the ORM and plugin work. Transport is resolved per resource/group +id by the same ``resolve_transport`` gate the execution path uses — the PG +queue where that is enabled, Celery otherwise. + +The group feature sits behind its own Flipt flag, evaluated once here at +enqueue: a blind Flipt, a missing org, or any dispatch error means no +notification, never a broken share. Turning the flag off stops new enqueues; a +message already queued still delivers. Direct-user share and revoke mail +(``ResourceShareManagementMixin``) is not on this flag — like the co-owner mail +it reuses, it is gated only by the cloud ``ENABLE_EMAIL_NOTIFICATIONS`` setting. """ from __future__ import annotations @@ -152,9 +155,14 @@ def _feature_enabled(organization_id: str) -> bool: # Parse exactly as FliptClient does (``.lower()``, no ``.strip()``) so the # two can never disagree on a value like " true". if os.environ.get("FLIPT_SERVICE_AVAILABLE", "false").lower() != "true": + logger.warning( + "group-notification: FLIPT_SERVICE_AVAILABLE != true (Flipt blind) " + "for org %s; skipping", + organization_id, + ) return False try: - return bool( + enabled = bool( check_feature_flag_status( flag_key=GROUP_NOTIFICATION_FLAG_KEY, entity_id=organization_id, @@ -168,6 +176,9 @@ def _feature_enabled(organization_id: str) -> bool: exc_info=True, ) return False + if not enabled: + logger.info("group-notification: flag off for org %s; skipping", organization_id) + return enabled def _organization_slug(obj: Any) -> str | None: diff --git a/frontend/src/hooks/useCoOwnerManagement.jsx b/frontend/src/hooks/useCoOwnerManagement.jsx index c3fee93d10..ed00733da7 100644 --- a/frontend/src/hooks/useCoOwnerManagement.jsx +++ b/frontend/src/hooks/useCoOwnerManagement.jsx @@ -54,6 +54,8 @@ function useCoOwnerManagement({ service, setAlertDetails, onListRefresh }) { // branch) after the user has moved to a different resource. Mutation // callers pass the token captured BEFORE their POSTs so a modal switch // during the mutation itself is caught too, not just one mid-refresh. + // Returns true when the resource turned out to be gone, so the caller can + // leave that alert standing instead of overwriting it with its own. async (resourceId, requestId = latestRequestRef.current) => { try { const res = await service.getSharedUsers(resourceId); @@ -69,7 +71,7 @@ function useCoOwnerManagement({ service, setAlertDetails, onListRefresh }) { content: "This resource is no longer accessible. It may have been removed or your access has been revoked.", }); - return; + return true; } setAlertDetails( handleException(err, "Unable to refresh co-owner data"), @@ -142,7 +144,12 @@ function useCoOwnerManagement({ service, setAlertDetails, onListRefresh }) { await run(addUsers, (id) => service.addCoOwner(resourceId, id)); await run(removeUsers, (id) => service.removeCoOwner(resourceId, id)); // Reconverge the modal on true server state regardless of partial outcome. - await refreshCoOwnerData(resourceId, requestId); + const gone = await refreshCoOwnerData(resourceId, requestId); + if (gone) { + // The refresh already closed the modal, refreshed the list and raised + // its own alert — an apply summary on top of it would only mislead. + return true; + } onListRefresh?.(); setAlertDetails( buildApplyAlert( diff --git a/workers/notification/tasks.py b/workers/notification/tasks.py index 2de8d237fd..1f8b60a3b4 100644 --- a/workers/notification/tasks.py +++ b/workers/notification/tasks.py @@ -489,7 +489,10 @@ def _post_group_notification(endpoint: str, organization_id: str, payload: dict) silently unsent email. On the PG transport the raise also leaves the message on the queue for redelivery, bounded by the consumer's attempt cap. - A 4xx is not retried — a rejected payload will be rejected again. + A sub-500 response ends the in-process attempts — a rejected payload will + be rejected again. Delivery is at-least-once: a response lost after the + backend already sent re-posts the same payload, and the send path writes + nothing, so the only effect is a duplicate email. """ base_url = os.getenv("INTERNAL_API_BASE_URL") api_key = os.getenv("INTERNAL_SERVICE_API_KEY") @@ -510,7 +513,7 @@ def _post_group_notification(endpoint: str, organization_id: str, payload: dict) try: with httpx.Client(transport=httpx.HTTPTransport(retries=2)) as client: response = client.post(url, headers=headers, json=payload, timeout=30.0) - except Exception as e: # noqa: BLE001 - transport failure, retry below + except Exception as e: # noqa: BLE001 last_error = f"exception={e!r}" else: if response.status_code == 200: From 9e9f57c4addc13bc63706b196338b6a8bd77e40b Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Wed, 5 Aug 2026 18:39:33 +0530 Subject: [PATCH 06/10] UN-3494 [MISC] Drop the local-only worker-pg-notification compose service The PG-queue notification consumer is local dev config and does not belong in the PR. The k8s chart already carries workerPgNotification from UN-3445 (#1688), which is the real deployment surface. Co-Authored-By: Claude Opus 5 --- docker/docker-compose.yaml | 36 ------------------------------------ 1 file changed, 36 deletions(-) diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml index d74dc2bf43..9a8db90afe 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -827,42 +827,6 @@ services: profiles: - pg-queue - # Notification consumer — webhook POSTs and the group-share emails (UN-3494). - # Without this, a notification enqueued on PG is durably stored and never run. - # Every task here is one short outbound HTTP call, so it stays light. - worker-pg-notification: - image: unstract/worker-unified:${VERSION} - container_name: unstract-worker-pg-notification - restart: unless-stopped - command: ["pg-queue-consumer"] - ports: - - "8101:8090" - env_file: - - ../workers/.env - - ./essentials.env - depends_on: - - db - - redis - environment: - - ENVIRONMENT=development - - APPLICATION_NAME=unstract-worker-pg-notification - - WORKER_BARRIER_BACKEND=pg - - WORKER_PG_QUEUE_CONSUMER_WORKER_TYPE=notification - - WORKER_PG_QUEUE_CONSUMER_QUEUE=notifications - - WORKER_PG_QUEUE_CONSUMER_HEALTH_PORT=8090 - - WORKER_PG_QUEUE_CONSUMER_CONCURRENCY=${PG_NOTIFICATION_CONCURRENCY:-4} - # One internal-API call (30s timeout) plus a SendGrid batch for the - # largest group, with headroom. Health-stale sits at or above it. - - WORKER_PG_QUEUE_CONSUMER_VT_SECONDS=${PG_NOTIFICATION_VT_SECONDS:-120} - - WORKER_PG_QUEUE_CONSUMER_HEALTH_STALE_SECONDS=${PG_NOTIFICATION_HEALTH_STALE_SECONDS:-180} - labels: - - traefik.enable=false - volumes: - - ./workflow_data:/data - - ${TOOL_REGISTRY_CONFIG_SRC_PATH}:/data/tool_registry_config - profiles: - - pg-queue - # Reaper / orchestrator — leader-elected loop. Run exactly ONE instance (it # elects a single leader via pg_orchestrator_lock; extra replicas idle as # standby). Besides barrier-orphan recovery it runs the PG scheduler tick From c663694c02c001756b452d57e4d357c3022cbdad Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Wed, 5 Aug 2026 18:55:35 +0530 Subject: [PATCH 07/10] UN-3494 [FIX] Exclude members who joined a group after its access was revoked A revoke resolves recipients from the group's live membership at delivery time, so anyone who joined between the click and the send was told their access was removed for a group through which they never held it. Normally a few seconds; on the PG transport with no consumer deployed the backlog can sit far longer. The revoke now carries the timestamp of the change and delivery drops memberships created after it. One string on the payload rather than the frozen member list, which would grow with the group. Co-Authored-By: Claude Opus 5 --- .../group_notification_service.py | 24 +++++++++++++---- backend/tenant_account_v2/internal_views.py | 4 ++- .../tenant_account_v2/share_notifications.py | 27 +++++++++++++------ workers/notification/tasks.py | 5 +++- 4 files changed, 45 insertions(+), 15 deletions(-) diff --git a/backend/tenant_account_v2/group_notification_service.py b/backend/tenant_account_v2/group_notification_service.py index 7030d642f2..e16139617b 100644 --- a/backend/tenant_account_v2/group_notification_service.py +++ b/backend/tenant_account_v2/group_notification_service.py @@ -26,6 +26,7 @@ if TYPE_CHECKING: from collections.abc import Iterable + from datetime import datetime logger = logging.getLogger(__name__) @@ -73,11 +74,13 @@ def send_resource_shared( resource_kind: str, resource_id: str, share_action: str = ShareAction.SHARED.value, + revoked_at: datetime | None = None, ) -> None: """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. + the recipient actually belongs to. ``share_action`` picks the wording, and + on a revoke ``revoked_at`` bounds who counts as "current". """ service = _service() if service is None: @@ -96,7 +99,7 @@ def send_resource_shared( return retained = _retained_user_ids(shared.instance, share_action) for group in _groups_in_org(organization, group_ids): - recipients = _group_recipients(organization, group, retained) + recipients = _group_recipients(organization, group, retained, revoked_at) logger.info( "group-notification: task=notify_resource_shared_with_group " "group_id=%s action=%s recipient_count=%d", @@ -197,11 +200,22 @@ def _retained_user_ids(resource: Any, share_action: str) -> set[int]: def _group_recipients( - organization: Organization, group: OrganizationGroup, retained: set[int] + organization: Organization, + group: OrganizationGroup, + retained: set[int], + joined_before: datetime | None = None, ) -> list[User]: - """Live members of ``group`` who did not keep access via ``retained``.""" + """Live members of ``group`` who did not keep access via ``retained``. + + ``joined_before`` (a revoke's timestamp) 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. + """ + memberships = group.memberships + if joined_before is not None: + memberships = memberships.filter(created_at__lte=joined_before) users = _live_member_users( - organization, group.memberships.values_list("user_id", flat=True) + organization, memberships.values_list("user_id", flat=True) ) return [user for user in users if user.pk not in retained] diff --git a/backend/tenant_account_v2/internal_views.py b/backend/tenant_account_v2/internal_views.py index c5797b53c0..47c292580f 100644 --- a/backend/tenant_account_v2/internal_views.py +++ b/backend/tenant_account_v2/internal_views.py @@ -37,10 +37,12 @@ class ResourceSharedWithGroupSerializer(serializers.Serializer): actor_id = serializers.IntegerField() resource_kind = serializers.CharField() resource_id = serializers.CharField() - # Defaulted so messages enqueued before this field existed still validate. share_action = serializers.ChoiceField( choices=[a.value for a in ShareAction], default=ShareAction.SHARED.value ) + # 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, default=None) class GroupMembershipChangedSerializer(serializers.Serializer): diff --git a/backend/tenant_account_v2/share_notifications.py b/backend/tenant_account_v2/share_notifications.py index 50155b0aba..31d94dfec2 100644 --- a/backend/tenant_account_v2/share_notifications.py +++ b/backend/tenant_account_v2/share_notifications.py @@ -26,6 +26,8 @@ from enum import StrEnum from typing import TYPE_CHECKING, Any +from django.utils import timezone + from tenant_account_v2.shareable_resources import kind_for_instance from unstract.core.data_models import is_pg_transport from unstract.flags.feature_flag import check_feature_flag_status @@ -94,6 +96,12 @@ def _notify_group_share( 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 (see the PG rollout + ordering note). 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: @@ -102,16 +110,19 @@ def _notify_group_share( kind = kind_for_instance(resource) if not organization_id or kind is None or not _feature_enabled(organization_id): return + kwargs: dict[str, Any] = { + "group_ids": group_ids, + "actor_id": actor.pk, + "resource_kind": kind, + "resource_id": str(resource.pk), + "share_action": str(share_action), + "organization_id": organization_id, + } + if share_action is ShareAction.REVOKED: + kwargs["revoked_at"] = timezone.now().isoformat() _dispatch_quietly( task_name=NOTIFY_RESOURCE_SHARED_TASK, - kwargs={ - "group_ids": group_ids, - "actor_id": actor.pk, - "resource_kind": kind, - "resource_id": str(resource.pk), - "share_action": str(share_action), - "organization_id": organization_id, - }, + kwargs=kwargs, organization_id=organization_id, entity_id=str(resource.pk), ) diff --git a/workers/notification/tasks.py b/workers/notification/tasks.py index 1f8b60a3b4..a85ed0c2aa 100644 --- a/workers/notification/tasks.py +++ b/workers/notification/tasks.py @@ -541,10 +541,12 @@ def notify_resource_shared_with_group( resource_id: str, organization_id: str, share_action: str = "shared", + revoked_at: str | None = None, ) -> None: """Email every current member of the groups whose access just changed. - ``share_action`` defaults so messages enqueued before it existed still run. + ``revoked_at`` is set on a revoke only; the backend uses it to skip members + who joined the group after the access was taken away. """ _post_group_notification( "resource-shared", @@ -555,6 +557,7 @@ def notify_resource_shared_with_group( "resource_kind": resource_kind, "resource_id": resource_id, "share_action": share_action, + "revoked_at": revoked_at, }, ) From 6ab867b0cde7e1a3dd74e4483637431a5420e6f2 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Wed, 5 Aug 2026 19:00:55 +0530 Subject: [PATCH 08/10] UN-3494 [FIX] Skip a queued grant email when the group's access is already gone MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A grant enqueued before a revoke could still be delivered after it, mailing the resource name and id to members who can no longer reach the resource. Delivery now revalidates the live ResourceGroupShare on the grant direction and drops groups that no longer hold it. The revoke direction needs no equivalent check — its share row is gone by delivery, and _retained_user_ids already covers members who kept access another way. Co-Authored-By: Claude Opus 5 --- .../group_notification_service.py | 25 ++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/backend/tenant_account_v2/group_notification_service.py b/backend/tenant_account_v2/group_notification_service.py index e16139617b..e577e9359f 100644 --- a/backend/tenant_account_v2/group_notification_service.py +++ b/backend/tenant_account_v2/group_notification_service.py @@ -98,7 +98,7 @@ def send_resource_shared( ) return retained = _retained_user_ids(shared.instance, share_action) - for group in _groups_in_org(organization, group_ids): + for group in _groups_to_mail(organization, group_ids, shared.instance, share_action): recipients = _group_recipients(organization, group, retained, revoked_at) logger.info( "group-notification: task=notify_resource_shared_with_group " @@ -199,6 +199,29 @@ def _retained_user_ids(resource: Any, share_action: str) -> set[int]: } +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)} + return [group for group in groups if group.pk in live] + + def _group_recipients( organization: Organization, group: OrganizationGroup, From 70b42c8126c7e9cf435e6974427ba412a3217080 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Wed, 5 Aug 2026 19:44:33 +0530 Subject: [PATCH 09/10] UN-3494 [FIX] Stamp the revoke cutoff before the feature-flag round-trip revoked_at was captured after _feature_enabled(), so the window between the share-removal commit and the timestamp spanned a Flipt network call. A user joining the group inside it passed the cutoff and was mailed a revocation. Co-Authored-By: Claude Opus 5 --- backend/tenant_account_v2/share_notifications.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/backend/tenant_account_v2/share_notifications.py b/backend/tenant_account_v2/share_notifications.py index 31d94dfec2..f418be24e1 100644 --- a/backend/tenant_account_v2/share_notifications.py +++ b/backend/tenant_account_v2/share_notifications.py @@ -106,6 +106,11 @@ def _notify_group_share( group_ids = sorted(group.pk for group in groups) if not group_ids: return + # Stamped before the Flipt round-trip below: a member joining inside that + # window would be mailed a revocation for 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 or not _feature_enabled(organization_id): @@ -118,8 +123,8 @@ def _notify_group_share( "share_action": str(share_action), "organization_id": organization_id, } - if share_action is ShareAction.REVOKED: - kwargs["revoked_at"] = timezone.now().isoformat() + if revoked_at is not None: + kwargs["revoked_at"] = revoked_at _dispatch_quietly( task_name=NOTIFY_RESOURCE_SHARED_TASK, kwargs=kwargs, From 3f7016e8da967fff9596ab153a1922379cfb67d6 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Wed, 5 Aug 2026 20:26:29 +0530 Subject: [PATCH 10/10] UN-3494 [TEST] Cover share/revoke notifications for users and groups Enqueue side (unit tier, no DB): payload shape, the revoked_at stamp landing before the Flipt round-trip, and the skip/swallow paths. Delivery side (integration tier): recipient selection - the live re-read on a grant, the revoked_at cutoff, org scoping and retained access - plus the direct-user share/revoke wiring on the share endpoint. Co-Authored-By: Claude Opus 5 --- .../tests/test_share_notifications.py | 104 ++++++++++ .../test_share_notification_dispatch.py | 183 ++++++++++++++++++ backend/tenant_account_v2/tests.py | 123 +++++++++++- 3 files changed, 409 insertions(+), 1 deletion(-) create mode 100644 backend/permissions/tests/test_share_notifications.py create mode 100644 backend/tenant_account_v2/test_share_notification_dispatch.py 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/tenant_account_v2/test_share_notification_dispatch.py b/backend/tenant_account_v2/test_share_notification_dispatch.py new file mode 100644 index 0000000000..8958b00e5d --- /dev/null +++ b/backend/tenant_account_v2/test_share_notification_dispatch.py @@ -0,0 +1,183 @@ +"""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``, +``_feature_enabled`` and ``_dispatch`` — and these run in the rig's unit tier +with no Postgres. The transport itself is covered by ``pg_queue.tests`` and +``workflow_manager.workflow_v2.tests.test_transport``; the delivery side by +``ResourceShareNotificationTests`` in ``tenant_account_v2.tests``. +""" + +from __future__ import annotations + +from contextlib import contextmanager +from datetime import UTC, datetime, timedelta +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(*, enabled: bool = True, dispatch_raises: Exception | None = None): + """Patch the module's three outbound seams; yield the flag + dispatch mocks.""" + with ( + patch.object(sn, "kind_for_instance", return_value="workflow"), + patch.object(sn, "_feature_enabled", return_value=enabled) as flag, + patch.object(sn, "_dispatch", side_effect=dispatch_raises) as dispatch, + ): + yield flag, 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["entity_id"] == "wf-1" + 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_revoked_at_is_stamped_before_the_flipt_round_trip(self): + # Regression (PR #2224): the stamp used to sit below ``_feature_enabled``, + # whose Flipt call is a network round-trip. Someone joining the group + # inside that window is mailed a revocation for access never held. + clock = [datetime(2026, 1, 1, 12, 0, tzinfo=UTC)] + + def _flipt(_org: str) -> bool: + clock[0] += timedelta(seconds=5) # stand-in for the Flipt round-trip + return True + + with ( + patch.object(sn.timezone, "now", side_effect=lambda: clock[0]), + patch.object(sn, "kind_for_instance", return_value="workflow"), + patch.object(sn, "_feature_enabled", side_effect=_flipt), + patch.object(sn, "_dispatch") as dispatch, + ): + sn.notify_resource_group_share_changed( + resource=_RESOURCE, added=[], removed=[_group(3)], actor=_ACTOR + ) + revoked_at = dispatch.call_args.kwargs["kwargs"]["revoked_at"] + assert revoked_at == datetime(2026, 1, 1, 12, 0, tzinfo=UTC).isoformat() + + 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_before_the_flag_check(self): + with _seams() as (flag, dispatch): + sn.notify_resource_group_share_changed( + resource=_RESOURCE, added=[], removed=[], actor=_ACTOR + ) + dispatch.assert_not_called() + flag.assert_not_called() # no Flipt call for a no-op share + + def test_flag_off_skips_dispatch(self): + with _seams(enabled=False) as (_, dispatch): + sn.notify_resource_group_share_changed( + resource=_RESOURCE, added=[_group(1)], 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, "_feature_enabled", return_value=True), + 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_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["entity_id"] == "9" + 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_before_the_flag_check(self): + with _seams() as (flag, 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() + flag.assert_not_called() diff --git a/backend/tenant_account_v2/tests.py b/backend/tenant_account_v2/tests.py index a105593599..eef404389c 100644 --- a/backend/tenant_account_v2/tests.py +++ b/backend/tenant_account_v2/tests.py @@ -11,18 +11,24 @@ """ 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 ( + send_membership_changed, + send_resource_shared, +) from tenant_account_v2.group_views import OrganizationGroupViewSet from tenant_account_v2.models import ( GroupMembership, @@ -30,6 +36,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, @@ -482,3 +489,117 @@ 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, + ) -> None: + 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_group_from_another_org_is_never_mailed(self) -> None: + 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 + ) + for action in (ShareAction.SHARED.value, ShareAction.REVOKED.value): + self._send(group_ids=[foreign_group.id], share_action=action) + self.service.send_group_resource_shared_notification.assert_not_called() + + 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_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"] + )