Route share notifications through NotificationService

Add notifications_enabled and send_internal_notification on
NotificationService and route ShareableService through them, so
PagesService / WorkflowsService / VisualizationsService /
HistoriesService no longer have to inject NotificationManager
just to forward it.

lib/galaxy/webapps/galaxy/services/histories.py

@@ -40,7 +40,6 @@ from galaxy.managers.histories import (
     HistoryManager,
     HistorySerializer,
 )
-from galaxy.managers.notification import NotificationManager
 from galaxy.managers.users import UserManager
 from galaxy.model import HistoryDatasetAssociation
 from galaxy.model.scoped_session import galaxy_scoped_session
@@ -130,7 +129,6 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores):
         filters: HistoryFilters,
         short_term_storage_allocator: ShortTermStorageAllocator,
         notification_service: NotificationService,
-        notification_manager: NotificationManager,
     ):
         super().__init__(security)
         self.manager = manager
@@ -140,9 +138,7 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores):
         self.citations_manager = citations_manager
         self.history_export_manager = history_export_manager
         self.filters = filters
-        self.shareable_service = ShareableHistoryService(
-            self.manager, self.serializer, notification_service, notification_manager
-        )
+        self.shareable_service = ShareableHistoryService(self.manager, self.serializer, notification_service)
         self.short_term_storage_allocator = short_term_storage_allocator
 
     def index(

lib/galaxy/webapps/galaxy/services/notifications.py

@@ -48,6 +48,20 @@ class NotificationService(ServiceBase):
         self.notification_manager = notification_manager
         self.sse_manager = sse_manager
 
+    @property
+    def notifications_enabled(self) -> bool:
+        return self.notification_manager.notifications_enabled
+
+    def send_internal_notification(
+        self, request: NotificationCreateRequest, force_sync: bool = False
+    ) -> NotificationCreatedResponse:
+        """Send a system-emitted notification on behalf of internal callers (e.g. share flows).
+
+        Unlike :meth:`send_notification`, this skips admin/permission checks because the
+        caller has already resolved the recipient set and is not acting on user input.
+        """
+        return self.notification_manager.send_notification_internal(request, force_sync=force_sync)
+
     def send_notification(
         self, sender_context: ProvidesUserContext, payload: NotificationCreateRequestBody
     ) -> Union[NotificationCreatedResponse, AsyncTaskResultSummary]:

lib/galaxy/webapps/galaxy/services/pages.py

@@ -11,7 +11,6 @@ from galaxy.managers.markdown_util import (
     internal_galaxy_markdown_to_pdf,
     to_basic_markdown,
 )
-from galaxy.managers.notification import NotificationManager
 from galaxy.managers.pages import (
     PageManager,
     PageSerializer,
@@ -59,14 +58,11 @@ class PagesService(ServiceBase):
         serializer: PageSerializer,
         short_term_storage_allocator: ShortTermStorageAllocator,
         notification_service: NotificationService,
-        notification_manager: NotificationManager,
     ):
         super().__init__(security)
         self.manager = manager
         self.serializer = serializer
-        self.shareable_service = ShareableService(
-            self.manager, self.serializer, notification_service, notification_manager
-        )
+        self.shareable_service = ShareableService(self.manager, self.serializer, notification_service)
         self.short_term_storage_allocator = short_term_storage_allocator
 
     def index(

lib/galaxy/webapps/galaxy/services/sharable.py

@@ -5,7 +5,6 @@ from typing import (
 )
 
 from galaxy.managers import base
-from galaxy.managers.notification import NotificationManager
 from galaxy.managers.sharable import (
     SharableModelManager,
     SharableModelSerializer,
@@ -63,12 +62,10 @@ class ShareableService:
         manager: SharableModelManager,
         serializer: SharableModelSerializer,
         notification_service: NotificationService,
-        notification_manager: NotificationManager,
     ) -> None:
         self.manager = manager
         self.serializer = serializer
         self.notification_service = notification_service
-        self.notification_manager = notification_manager
 
     def set_slug(self, trans, id: DecodedDatabaseIdField, payload: SetSlugPayload):
         item = self._get_item_by_id(trans, id)
@@ -180,13 +177,13 @@ class ShareableService:
     def _send_notification_to_users(
         self, users_to_notify: set[User], item: SharableItem, status: ShareWithStatus, galaxy_url: Optional[str] = None
     ):
-        if self.notification_manager.notifications_enabled and not status.errors and users_to_notify:
+        if self.notification_service.notifications_enabled and not status.errors and users_to_notify:
             request = SharedItemNotificationFactory.build_notification_request(
                 item, users_to_notify, status, galaxy_url
             )
             # We can set force_sync=True here because we already have the set of users to notify
             # and there is no need to resolve them asynchronously as no groups or roles are involved.
-            self.notification_manager.send_notification_internal(request, force_sync=True)
+            self.notification_service.send_internal_notification(request, force_sync=True)
 
 
 class SharedItemNotificationFactory:

lib/galaxy/webapps/galaxy/services/visualizations.py

@@ -9,7 +9,6 @@ from typing import (
 from galaxy import exceptions
 from galaxy.managers.base import security_check
 from galaxy.managers.context import ProvidesUserContext
-from galaxy.managers.notification import NotificationManager
 from galaxy.managers.sharable import (
     slug_exists,
     SlugBuilder,
@@ -63,14 +62,11 @@ class VisualizationsService(ServiceBase):
         manager: VisualizationManager,
         serializer: VisualizationSerializer,
         notification_service: NotificationService,
-        notification_manager: NotificationManager,
     ):
         super().__init__(security)
         self.manager = manager
         self.serializer = serializer
-        self.shareable_service = ShareableService(
-            self.manager, self.serializer, notification_service, notification_manager
-        )
+        self.shareable_service = ShareableService(self.manager, self.serializer, notification_service)
 
     # TODO: add the rest of the API actions here and call them directly from the API controller
 

lib/galaxy/webapps/galaxy/services/workflows.py

@@ -12,7 +12,6 @@ from galaxy import (
     web,
 )
 from galaxy.managers.context import ProvidesUserContext
-from galaxy.managers.notification import NotificationManager
 from galaxy.managers.workflows import (
     RefactorRequest,
     RefactorResponse,
@@ -58,14 +57,11 @@ class WorkflowsService(ServiceBase):
         serializer: WorkflowSerializer,
         tool_shed_registry: Registry,
         notification_service: NotificationService,
-        notification_manager: NotificationManager,
     ):
         self._workflows_manager = workflows_manager
         self._workflow_contents_manager = workflow_contents_manager
         self._serializer = serializer
-        self.shareable_service = ShareableService(
-            workflows_manager, serializer, notification_service, notification_manager
-        )
+        self.shareable_service = ShareableService(workflows_manager, serializer, notification_service)
         self._tool_shed_registry = tool_shed_registry
 
     def index(