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/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/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 1227562531..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) - - @property - def added(self) -> set[Any]: - return self.after - self.before +def _notification_context(view: Any, instance: Any) -> tuple[str, str] | None: + """Resolve ``(resource_type, resource_name)`` for the email senders. - @property - def removed(self) -> set[Any]: - return self.before - self.after + ``None`` when the plugin is absent or the host ViewSet has not opted in by + setting ``notification_resource_name_field`` and overriding + ``get_notification_resource_type`` (both declared on + ``OwnerManagementMixin``, which every share host also mixes in). + """ + name_field = getattr(view, "notification_resource_name_field", None) + resolve_type = getattr(view, "get_notification_resource_type", None) + if not notification_plugin or not name_field or resolve_type is None: + return None + resource_type = resolve_type(instance) + resource_name = getattr(instance, name_field, None) + if resource_type is None or not resource_name: + return None + return resource_type, resource_name + + +def _users_left_without_access(instance: Model, users: set[Any]) -> list[Any]: + """Narrow ``users`` to those with no remaining access to ``instance``. + + Someone dropped from ``shared_users`` may still reach the resource via a + group or an org-wide share; telling them their access was removed would be + wrong. + """ + if not users: + return [] + from tenant_account_v2.sharing_helpers import compute_effective_members + + retained = {member["user_id"] for member in compute_effective_members(instance)} + return [user for user in users if user.pk not in retained] + + +def _send_share_notification( + instance: Model, context: tuple[str, str], users: set[Any], actor: Any +) -> None: + """Email users newly granted direct access. Best-effort.""" + resource_type, resource_name = context + try: + notification_plugin["service_class"]().send_sharing_notification( + resource_type=resource_type, + resource_name=resource_name, + resource_id=str(instance.pk), + shared_by=actor, + shared_to=list(users), + resource_instance=instance, + ) + except Exception: + logger.exception("Failed to send sharing notification for %s", instance.pk) + + +def _send_revoke_notification( + instance: Model, context: tuple[str, str], users: list[Any], actor: Any +) -> None: + """Email users whose direct access was revoked. Best-effort.""" + resource_type, resource_name = context + try: + notification_plugin["service_class"]().send_access_removed_notification( + resource_type=resource_type, + resource_name=resource_name, + resource_id=str(instance.pk), + removed_from=users, + removed_by=actor, + resource_instance=instance, + ) + except Exception: + logger.exception("Failed to send access-removed notification for %s", instance.pk) class ResourceShareManagementMixin: - """Adds the shared share-management surface to a resource ViewSet. - - Subclasses declare share axes via :attr:`share_axes`. The default - covers ``shared_users`` + ``shared_groups``. - """ - - share_axes: ClassVar[tuple[str, ...]] = ("shared_users", "shared_groups") + """Adds the shared share-management surface to a resource ViewSet.""" @action(detail=True, methods=["post"], url_path="share") def share(self, request: Request, pk: str | None = None) -> Response: @@ -91,15 +143,57 @@ def share(self, request: Request, pk: str | None = None) -> Response: users, group-membership for groups) live in ``ShareAuthorizationService``. """ + from tenant_account_v2.share_notifications import ( + notify_resource_group_share_changed, + ) from tenant_account_v2.sharing_helpers import ShareAuthorizationService resource = self.get_object() # type: ignore[attr-defined] desired = _extract_desired_share_state(request.data) + users_before = self._read_axis(resource, "shared_users") + groups_before = self._read_axis(resource, "shared_groups") ShareAuthorizationService.authorize_and_commit( actor=request.user, resource=resource, desired=desired ) + # ``_commit`` is the only atomic block on this path, so it has already + # committed — the diffs read persisted state and can never announce a + # share that rolled back. + resource.refresh_from_db() + users_after = self._read_axis(resource, "shared_users") + groups_after = self._read_axis(resource, "shared_groups") + notify_resource_group_share_changed( + resource=resource, + added=groups_after - groups_before, + removed=groups_before - groups_after, + actor=request.user, + ) + self._notify_shared_users( + resource, users_after - users_before, users_before - users_after, request.user + ) return Response(status=status.HTTP_200_OK) + def _notify_shared_users( + self, + instance: Any, + added: set[Any], + removed: set[Any], + actor: Any, + /, + ) -> None: + """Email users granted or denied direct access. + + Resource type and name come from the host's ``OwnerManagementMixin`` + seam, so every share host is covered without an override. + """ + context = _notification_context(self, instance) + if context is None: + return + if added: + _send_share_notification(instance, context, added, actor) + revoked = _users_left_without_access(instance, removed) + if revoked: + _send_revoke_notification(instance, context, revoked, actor) + @action(detail=True, methods=["get"], url_path="effective-members") def effective_members(self, request: Request, pk: str | None = None) -> Response: """Return all users with access (direct/group/org), priority-deduped.""" @@ -113,36 +207,6 @@ def effective_members(self, request: Request, pk: str | None = None) -> Response members = compute_effective_members(self.get_object()) # type: ignore[attr-defined] return Response(EffectiveMemberSerializer(members, many=True).data) - def snapshot_share_axes(self, instance: Model) -> dict[str, set[Any]]: - """Capture every declared axis's current contents. - - Call BEFORE ``super().partial_update(...)``; pair with - :meth:`diff_share_axes` afterward. - """ - return {axis: self._read_axis(instance, axis) for axis in self.share_axes} - - def diff_share_axes( - self, - instance: Model, - before: dict[str, set[Any]], - request_data: dict[str, Any], - ) -> dict[str, AxisDiff]: - """Diff each axis that was touched by the request. - - Returns a dict keyed by axis name with only the axes present in - ``request_data`` — callers can skip notification fan-out for axes - the client did not modify. - """ - instance.refresh_from_db() - return { - axis: AxisDiff( - before=before[axis], - after=self._read_axis(instance, axis), - ) - for axis in self.share_axes - if axis in request_data - } - @staticmethod def _read_axis(instance: Model, axis: str) -> set[Any]: """Return the current set of related objects on the given axis. 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 new file mode 100644 index 0000000000..ccdc3d9989 --- /dev/null +++ b/backend/tenant_account_v2/group_notification_service.py @@ -0,0 +1,240 @@ +"""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, ShareAction +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, + share_action: str = ShareAction.SHARED.value, +) -> 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. + """ + 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 action=%s recipient_count=%d", + "notify_resource_shared_with_group", + 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, + ) + + +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..0e959c9b5d --- /dev/null +++ b/backend/tenant_account_v2/internal_views.py @@ -0,0 +1,95 @@ +"""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, ShareAction + +logger = logging.getLogger(__name__) + + +class ResourceSharedWithGroupSerializer(serializers.Serializer): + """Payload of ``notify_resource_shared_with_group``.""" + + group_ids = serializers.ListField(child=serializers.IntegerField(), allow_empty=False) + actor_id = serializers.IntegerField() + resource_kind = serializers.CharField() + resource_id = serializers.CharField() + # 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): + """Payload of ``notify_group_membership_changed``.""" + + group_id = serializers.IntegerField() + actor_id = serializers.IntegerField() + membership_action = serializers.ChoiceField( + choices=[a.value for a in MembershipAction] + ) + user_ids = serializers.ListField(child=serializers.IntegerField(), allow_empty=False) + + +class _GroupNotificationView(APIView): + """Shared org resolution for the group-notification endpoints.""" + + @staticmethod + def _organization() -> Organization: + organization = UserContext.get_organization() + if organization is None: + raise ValidationError( + "Organization context missing. Worker must send X-Organization-ID." + ) + return organization + + +class ResourceSharedWithGroupView(_GroupNotificationView): + """Mail every current member of the groups whose resource access just changed.""" + + def post(self, request: Request) -> Response: + serializer = ResourceSharedWithGroupSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + data = serializer.validated_data + try: + 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..526f926a04 --- /dev/null +++ b/backend/tenant_account_v2/share_notifications.py @@ -0,0 +1,241 @@ +"""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 or revoked +from the group, and a user added to or removed from it. Both are dispatched +asynchronously — the caller's request returns as soon as the write lands. + +The sending itself runs in ``workers/``, which is Django-free, so the worker +task is a thin HTTP shim back to :mod:`tenant_account_v2.internal_views`; the +backend does the ORM and plugin work. Transport is 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" + + +class ShareAction(StrEnum): + """What happened to a group's access to a resource.""" + + SHARED = "shared" + REVOKED = "revoked" + + +def notify_resource_group_share_changed( + *, + resource: Any, + added: Iterable[OrganizationGroup], + removed: Iterable[OrganizationGroup], + actor: User, +) -> None: + """Queue group mail for a resource just shared with / revoked from groups.""" + for share_action, groups in ( + (ShareAction.SHARED, added), + (ShareAction.REVOKED, removed), + ): + _notify_group_share( + resource=resource, groups=groups, share_action=share_action, actor=actor + ) + + +def _notify_group_share( + *, + resource: Any, + groups: Iterable[OrganizationGroup], + share_action: ShareAction, + actor: User, +) -> None: + """Queue one group-share event. + + Recipients are resolved at delivery time rather than frozen here: anyone + who leaves the org between the click and the send simply isn't in the fresh + lookup, so offboarding safety costs nothing. Unlike a membership removal, + revoking a group's access leaves the group and its members intact, so the + fresh lookup still finds everyone who needs telling. + """ + 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), + "share_action": str(share_action), + "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/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/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/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, }; } diff --git a/workers/notification/tasks.py b/workers/notification/tasks.py index 41d76f5cf9..10945e34c8 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,109 @@ 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, + share_action: str = "shared", +) -> None: + """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, + { + "group_ids": group_ids, + "actor_id": actor_id, + "resource_kind": resource_kind, + "resource_id": resource_id, + "share_action": share_action, + }, + ) + + +@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."""