From 798438d7a64826976e8d543f5a011e780a71f631 Mon Sep 17 00:00:00 2001 From: kirtimanmishrazipstack Date: Fri, 31 Jul 2026 20:41:31 +0530 Subject: [PATCH 1/4] 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 2/4] 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 3/4] 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 4/4] 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, }; }