Issue1248: Persistent Email Reminders

backend/dataall/modules/shares_base/handlers/ecs_share_handler.py

@@ -30,6 +30,13 @@ def verify_share(engine, task: Task):
     def reapply_share(engine, task: Task):
         return EcsShareHandler._manage_share(engine, task, SharingService.reapply_share, 'reapply_share')
 
+    @staticmethod
+    @Worker.handler(path='ecs.share.persistent_email_reminder')
+    def persistent_email_reminder(engine, task: Task):
+        return EcsShareHandler._manage_share(
+            engine, task, SharingService.persistent_email_reminder, 'persistent_email_reminder'
+        )
+
     @staticmethod
     def _manage_share(engine, task: Task, local_handler, ecs_handler: str):
         envname = os.environ.get('envname', 'local')

backend/dataall/modules/shares_base/services/share_notification_service.py

@@ -56,11 +56,50 @@ def notify_share_object_submission(self, email_id: str):
         self._create_notification_task(subject=subject, msg=email_notification_msg)
         return notifications
 
+    def notify_persistent_email_reminder(self, email_id: str, engine):
+        share_link_text = ''
+        if os.environ.get('frontend_domain_url'):
+            share_link_text = (
+                f'<br><br>Please visit data.all <a href="{os.environ.get("frontend_domain_url")}'
+                f'/console/shares/{self.share.shareUri}">share link</a> '
+                f'to review and take appropriate action or view more details.'
+            )
+
+        msg_intro = f"""Dear User,
+
+        This is a reminder that a share request for the dataset "{self.dataset.label}" submitted by {email_id} 
+        on behalf of principal "{self.share.principalId}" is still pending and has not been addressed.
+
+        """
+
+        msg_end = """Your prompt attention to this matter is greatly appreciated.
+
+        Best regards,
+        The Data.all Team
+        """
+
+        subject = f'URGENT REMINDER: Data.all | Action Required on Pending Share Request for {self.dataset.label}'
+        email_notification_msg = msg_intro + share_link_text + msg_end
+
+        notifications = self._register_notifications(
+            notification_type=DataSharingNotificationType.SHARE_OBJECT_SUBMITTED.value, msg=msg_intro
+        )
+
+        self._create_persistent_reminder_notification_task(subject=subject, msg=email_notification_msg, engine=engine)
+        return notifications
+
     def notify_share_object_approval(self, email_id: str):
         share_link_text = ''
         if os.environ.get('frontend_domain_url'):
-            share_link_text = f'<br><br> Please visit data.all <a href="{os.environ.get("frontend_domain_url")}/console/shares/{self.share.shareUri}">share link </a> to take action or view more details'
-        msg = f'User {email_id} APPROVED share request for dataset {self.dataset.label} for principal {self.share.principalId}'
+            share_link_text = (
+                f'<br><br> Please visit data.all <a href="{os.environ.get("frontend_domain_url")}'
+                f'/console/shares/{self.share.shareUri}">share link </a> '
+                f'to take action or view more details'
+            )
+        msg = (
+            f'User {email_id} APPROVED share request for dataset {self.dataset.label} '
+            f'for principal {self.share.principalId}'
+        )
         subject = f'Data.all | Share Request Approved for {self.dataset.label}'
         email_notification_msg = msg + share_link_text
 
@@ -175,3 +214,40 @@ def _create_notification_task(self, subject, msg):
                     log.info(f'Notification type : {share_notification_config_type} is not active')
         else:
             log.info('Notifications are not active')
+
+    def _create_persistent_reminder_notification_task(self, subject, msg, engine):
+        """
+        At the moment just for notification_config_type = email, but designed for additional notification types
+        Emails sent to:
+            - dataset.SamlAdminGroupName
+            - dataset.stewards
+        """
+        share_notification_config = config.get_property(
+            'modules.datasets_base.features.share_notifications', default=None
+        )
+        if share_notification_config:
+            for share_notification_config_type in share_notification_config.keys():
+                n_config = share_notification_config[share_notification_config_type]
+                if n_config.get('active', False) == True:
+                    notification_recipient_groups_list = [self.dataset.SamlAdminGroupName, self.dataset.stewards]
+
+                    if share_notification_config_type == 'email':
+                        notification_task: Task = Task(
+                            action='notification.service',
+                            targetUri=self.share.shareUri,
+                            payload={
+                                'notificationType': share_notification_config_type,
+                                'subject': subject,
+                                'message': msg,
+                                'recipientGroupsList': notification_recipient_groups_list,
+                                'recipientEmailList': [],
+                            },
+                        )
+                        self.session.add(notification_task)
+                        self.session.commit()
+
+                        Worker.queue(engine=engine, task_ids=[notification_task.taskUri])
+                else:
+                    log.info(f'Notification type : {share_notification_config_type} is not active')
+        else:
+            log.info('Notifications are not active')

backend/dataall/modules/shares_base/services/sharing_service.py

@@ -1,9 +1,10 @@
 import logging
 from typing import Any
 from dataclasses import dataclass
-from sqlalchemy import and_
+from abc import ABC, abstractmethod
+from sqlalchemy import and_, func
 from time import sleep
-from dataall.base.db import Engine
+from dataall.base.db import Engine, get_engine
 from dataall.core.environment.db.environment_models import Environment, EnvironmentGroup
 from dataall.modules.shares_base.db.share_object_state_machines import (
     ShareObjectSM,
@@ -17,8 +18,13 @@
 )
 from dataall.modules.shares_base.db.share_object_models import ShareObject
 from dataall.modules.shares_base.db.share_object_repositories import ShareObjectRepository
+
+from dataall.modules.shares_base.services.share_notification_service import ShareNotificationService
+from dataall.modules.s3_datasets.db.dataset_repositories import DatasetRepository
+from dataall.modules.notifications.db.notification_models import Notification
 from dataall.modules.shares_base.db.share_state_machines_repositories import ShareStatusRepository
 from dataall.modules.shares_base.services.share_processor_manager import ShareProcessorManager
+
 from dataall.modules.shares_base.services.share_object_service import (
     ShareObjectService,
 )
@@ -356,6 +362,48 @@ def reapply_share(cls, engine: Engine, share_uri: str) -> bool:
                     if not lock_released:
                         log.error(f'Failed to release lock for dataset {share_data.dataset.datasetUri}.')
 
+    @staticmethod
+    def fetch_pending_shares(engine):
+        """
+        A method used by the scheduled ECS Task to run fetch_pending_shares() process against ALL shared objects in ALL
+        active share objects within dataall
+        """
+        with engine.scoped_session() as session:
+            pending_shares = (
+                session.query(ShareObject)
+                .join(
+                    Notification,
+                    and_(
+                        ShareObject.shareUri == func.split_part(Notification.target_uri, '|', 1),
+                        ShareObject.datasetUri == func.split_part(Notification.target_uri, '|', 2),
+                    ),
+                )
+                .filter(and_(Notification.type == 'SHARE_OBJECT_SUBMITTED', ShareObject.status == 'Submitted'))
+                .all()
+            )
+            return pending_shares
+
+    @staticmethod
+    def _get_share_data(session, uri):
+        share = ShareObjectRepository.get_share_by_uri(session, uri)
+        dataset = DatasetRepository.get_dataset_by_uri(session, share.datasetUri)
+        share_items_states = ShareObjectRepository.get_share_items_states(session, uri)
+        return share, dataset, share_items_states
+
+    @classmethod
+    def persistent_email_reminder(cls, uri: str, envname):
+        """
+        A method used by the scheduled ECS Task to send email notifications to the requestor of the share object
+        """
+        engine = get_engine(envname=envname)
+
+        with engine.scoped_session() as session:
+            share, dataset, states = cls._get_share_data(session, uri)
+            ShareNotificationService(session=session, dataset=dataset, share=share).notify_persistent_email_reminder(
+                email_id=share.owner, engine=engine
+            )
+            log.info(f'Email reminder sent for share {share.shareUri}')
+
     @staticmethod
     def _get_share_data_and_items(session, share_uri, status, healthStatus=None):
         data = ShareObjectRepository.get_share_data(session, share_uri)

backend/dataall/modules/shares_base/tasks/persistent_email_reminders_task.py

@@ -0,0 +1,37 @@
+import logging
+import os
+import sys
+from dataall.modules.dataset_sharing.api.types import ShareObject
+from dataall.modules.shares_base.services.sharing_service import SharingService
+from dataall.base.db import get_engine
+from dataall.base.aws.sqs import SqsQueue
+from dataall.core.tasks.service_handlers import Worker
+
+root = logging.getLogger()
+root.setLevel(logging.INFO)
+if not root.hasHandlers():
+    root.addHandler(logging.StreamHandler(sys.stdout))
+log = logging.getLogger(__name__)
+Worker.queue = SqsQueue.send
+
+
+def persistent_email_reminders(engine, envname):
+    """
+    A method used by the scheduled ECS Task to run persistent_email_reminder() process against ALL
+    active share objects within data.all and send emails to all pending shares.
+    """
+    with engine.scoped_session() as session:
+        log.info('Running Persistent Email Reminders Task')
+        pending_shares = SharingService.fetch_pending_shares(engine=engine)
+        log.info(f'Found {len(pending_shares)} pending shares')
+        pending_share: ShareObject
+        for pending_share in pending_shares:
+            log.info(f'Sending Email Reminder for Share: {pending_share.shareUri}')
+            SharingService.persistent_email_reminder(uri=pending_share.shareUri, envname=envname)
+        log.info('Completed Persistent Email Reminders Task')
+
+
+if __name__ == '__main__':
+    ENVNAME = os.environ.get('envname', 'local')
+    ENGINE = get_engine(envname=ENVNAME)
+    persistent_email_reminders(engine=ENGINE, envname=ENVNAME)

config.json

@@ -18,6 +18,7 @@
                 "share_notifications": {
                     "email": {
                         "active": false,
+                        "persistent_reminders": false,
                         "parameters": {
                             "group_notifications": true
                         }

deploy/stacks/backend_stack.py

@@ -218,6 +218,9 @@ def __init__(
                 self.lambda_api_stack.api_handler,
                 self.lambda_api_stack.elasticsearch_proxy_handler,
             ],
+            email_custom_domain=ses_stack.ses_identity.email_identity_name if ses_stack is not None else None,
+            ses_configuration_set=ses_stack.configuration_set.configuration_set_name if ses_stack is not None else None,
+            custom_domain=custom_domain,
             **kwargs,
         )
 

deploy/stacks/container.py

@@ -31,6 +31,9 @@ def __init__(
         tooling_account_id=None,
         s3_prefix_list=None,
         lambdas=None,
+        email_custom_domain=None,
+        ses_configuration_set=None,
+        custom_domain=None,
         **kwargs,
     ):
         super().__init__(scope, id, **kwargs)
@@ -44,11 +47,20 @@ def __init__(
         self._ecr_repository = ecr_repository
         self._vpc = vpc
         self._prod_sizing = prod_sizing
+        self._email_custom_domain = email_custom_domain
+        self._ses_configuration_set = ses_configuration_set
 
         (self.scheduled_tasks_sg, self.share_manager_sg) = self.create_ecs_security_groups(
             envname, resource_prefix, vpc, vpce_connection, s3_prefix_list, lambdas
         )
         self.ecs_security_groups: [aws_ec2.SecurityGroup] = [self.scheduled_tasks_sg, self.share_manager_sg]
+        self.env_vars = self._create_env('INFO')
+
+        # Check if custom domain exists and if it exists email notifications could be enabled.
+        # Create an env variable which stores the domain URL.
+        # This is used for sending data.all share weblinks in the email notifications.
+        if custom_domain and custom_domain.get('hosted_zone_name'):
+            self.env_vars.update({'frontend_domain_url': f'https://{custom_domain["hosted_zone_name"]}'})
 
         cluster = ecs.Cluster(
             self,
@@ -58,7 +70,10 @@ def __init__(
             container_insights=True,
         )
 
-        self.task_role = self.create_task_role(envname, resource_prefix, pivot_role_name)
+        self.task_role = self.create_task_role(
+            envname, resource_prefix, pivot_role_name, email_custom_domain, ses_configuration_set
+        )
+
         self.cicd_stacks_updater_role = self.create_cicd_stacks_updater_role(
             envname, resource_prefix, tooling_account_id
         )
@@ -178,6 +193,7 @@ def __init__(
         self.add_share_verifier_task()
         self.add_share_reapplier_task()
         self.add_omics_fetch_workflows_task()
+        self.add_persistent_email_reminders_task()
 
     @run_if(['modules.s3_datasets.active', 'modules.dashboards.active'])
     def add_catalog_indexer_task(self):
@@ -286,6 +302,32 @@ def add_share_reapplier_task(self):
         )
         self.ecs_task_definitions_families.append(share_reapplier_task_definition.family)
 
+    @run_if(['modules.dataset_base.features.share_notifications.email.persistent_reminders'])
+    def add_persistent_email_reminders_task(self):
+        persistent_email_reminders_task, persistent_email_reminders_task_def = self.set_scheduled_task(
+            cluster=self.ecs_cluster,
+            command=[
+                'python3.9',
+                '-m',
+                'dataall.modules.shares_base.tasks.persistent_email_reminders_task',
+            ],
+            container_id='container',
+            ecr_repository=self._ecr_repository,
+            environment=self.env_vars,
+            image_tag=self._cdkproxy_image_tag,
+            log_group=self.create_log_group(
+                self._envname, self._resource_prefix, log_group_name='persistent-email-reminders'
+            ),
+            schedule_expression=Schedule.expression('cron(0 9 ? * 2 *)'),  # Run at 9:00 AM UTC every Monday
+            scheduled_task_id=f'{self._resource_prefix}-{self._envname}-persistent-email-reminders-schedule',
+            task_id=f'{self._resource_prefix}-{self._envname}-persistent-email-reminders',
+            task_role=self.task_role,
+            vpc=self._vpc,
+            security_group=self.scheduled_tasks_sg,
+            prod_sizing=self._prod_sizing,
+        )
+        self.ecs_task_definitions_families.append(persistent_email_reminders_task.task_definition.family)
+
     @run_if(['modules.s3_datasets.active'])
     def add_subscription_task(self):
         subscriptions_task, subscription_task_def = self.set_scheduled_task(
@@ -453,7 +495,9 @@ def create_cicd_stacks_updater_role(self, envname, resource_prefix, tooling_acco
         )
         return cicd_stacks_updater_role
 
-    def create_task_role(self, envname, resource_prefix, pivot_role_name):
+    def create_task_role(
+        self, envname, resource_prefix, pivot_role_name, email_custom_domain=None, ses_configuration_set=None
+    ):
         role_inline_policy = iam.Policy(
             self,
             f'ECSRolePolicy{envname}',
@@ -542,13 +586,26 @@ def create_task_role(self, envname, resource_prefix, pivot_role_name):
                 ),
             ],
         )
+
+        if email_custom_domain and ses_configuration_set:
+            role_inline_policy.document.add_statements(
+                iam.PolicyStatement(
+                    actions=['ses:SendEmail'],
+                    resources=[
+                        f'arn:aws:ses:{self.region}:{self.account}:identity/{email_custom_domain}',
+                        f'arn:aws:ses:{self.region}:{self.account}:configuration-set/{ses_configuration_set}',
+                    ],
+                )
+            )
+
         task_role = iam.Role(
             self,
             f'ECSTaskRole{envname}',
             role_name=f'{resource_prefix}-{envname}-ecs-tasks-role',
             inline_policies={f'ECSRoleInlinePolicy{envname}': role_inline_policy.document},
             assumed_by=iam.ServicePrincipal('ecs-tasks.amazonaws.com'),
         )
+
         task_role.grant_pass_role(task_role)
         return task_role