HDX-10154 push unsubscribe token to Novu

- for both new subscribers and existing ones
- also check for existing tokens before creating new ones (otherwise you get errors when a user with the same email tries to subscribe to the same dataset twice in a row)

ckanext-hdx_users/ckanext/hdx_users/actions/update.py

@@ -131,6 +131,8 @@ def hdx_add_notification_subscription(context: Context, data_dict: DataDict):
 
     email = data_dict.get('email')
     dataset_id = data_dict.get('dataset_id')
+    unsubscribe_token = data_dict.get('unsubscribe_token')
+    unsubscribe_token_key = 'unsubscribeToken_' + dataset_id.replace('-', '_')
 
     if not email or not dataset_id:
         raise tk.ValidationError('Missing required parameters: email and dataset_id')
@@ -153,13 +155,25 @@ def hdx_add_notification_subscription(context: Context, data_dict: DataDict):
         # Subscriber doesn't exist; create a new one
         subscriber_data = {
             'subscriberId': subscriber_id,
-            'email': email
+            'email': email,
+            'data': {
+                unsubscribe_token_key: unsubscribe_token
+            }
         }
         response = requests.post(f'{novu_api_url}/subscribers', json=subscriber_data, headers=headers)
         if response.status_code != 201:
             raise Exception(f'Failed to create subscriber: {response.text}')
 
-    elif response.status_code != 200:
+    elif response.status_code == 200:
+        data = response.json().get('data', {}).get('data', {})
+        data[unsubscribe_token_key] = unsubscribe_token
+        subscriber_data = {
+            'data': data
+        }
+        response = requests.put(f'{novu_api_url}/subscribers/{email}', json=subscriber_data, headers=headers)
+        if response.status_code != 200:
+            raise Exception(f'Failed to update subscriber: {response.text}')
+    else:
         raise Exception(f'Error checking subscriber: {response.text}')
 
     topic_key = f'dataset-{dataset_id}'

ckanext-hdx_users/ckanext/hdx_users/controller_logic/notification_platform_logic.py

@@ -4,7 +4,7 @@
 import ckan.model as model
 
 from ckanext.hdx_users.general_token_model import generate_new_token_obj, validate_token, ObjectType, TokenType, \
-    HDXGeneralToken
+    HDXGeneralToken, get_by_type_and_user_id_and_object
 from ckanext.hdx_users.helpers.notification_platform import check_notifications_enabled_for_dataset
 
 log = logging.getLogger(__name__)
@@ -14,18 +14,28 @@
 get_action = tk.get_action
 
 
-def generate_email_validation_token(email: str, dataset_id: str) -> HDXGeneralToken:
+def get_or_generate_email_validation_token(email: str, dataset_id: str) -> HDXGeneralToken:
     dataset_supports_notifications = check_notifications_enabled_for_dataset(dataset_id)
     if dataset_supports_notifications:
-        return generate_new_token_obj(model.Session, TokenType.EMAIL_VALIDATION_FOR_DATASET, email,
+        email_validation_token = get_by_type_and_user_id_and_object(TokenType.EMAIL_VALIDATION_FOR_DATASET, email,
+                                                                        ObjectType.DATASET, dataset_id)
+        if email_validation_token:
+            return email_validation_token
+        else:
+            return generate_new_token_obj(model.Session, TokenType.EMAIL_VALIDATION_FOR_DATASET, email,
                                       object_type=ObjectType.DATASET, object_id=dataset_id)
     else:
         log.warning(f'Tried to generate token for dataset {dataset_id} but dataset does not support notifications')
         raise Exception(f'Dataset {dataset_id} does not support notifications')
 
 
-def generate_unsubscribe_token(email: str, dataset_id: str) -> HDXGeneralToken:
-    return generate_new_token_obj(model.Session, TokenType.UNSUBSCRIBE_FOR_DATASET, email,
+def get_or_generate_unsubscribe_token(email: str, dataset_id: str) -> HDXGeneralToken:
+    existing_unsubscribe_token = get_by_type_and_user_id_and_object(TokenType.UNSUBSCRIBE_FOR_DATASET, email,
+                                                                    ObjectType.DATASET, dataset_id)
+    if existing_unsubscribe_token:
+        return existing_unsubscribe_token
+    else:
+        return generate_new_token_obj(model.Session, TokenType.UNSUBSCRIBE_FOR_DATASET, email,
                                   object_type=ObjectType.DATASET, object_id=dataset_id)
 
 

ckanext-hdx_users/ckanext/hdx_users/general_token_model.py

@@ -90,16 +90,30 @@ def get_by_token(token: str) -> Optional[HDXGeneralToken]:
     return meta.Session.query(HDXGeneralToken).filter(HDXGeneralToken.token == token).first()
 
 
-def get_by_type_and_user_id(token_type: str, user_id: str) -> Optional[List[HDXGeneralToken]]:
+def get_by_type_and_user_id(token_type: TokenType, user_id: str) -> Optional[List[HDXGeneralToken]]:
     if not token_type or not user_id:
         return None
 
     return meta.Session.query(HDXGeneralToken) \
-            .filter(HDXGeneralToken.token_type == token_type) \
+            .filter(HDXGeneralToken.token_type == token_type.value) \
             .filter(HDXGeneralToken.user_id == user_id) \
             .all()
 
 
+def get_by_type_and_user_id_and_object(token_type: TokenType, user_id: str,
+                                       object_type: ObjectType, object_id: str) -> Optional[HDXGeneralToken]:
+    if not token_type or not user_id or not object_type or not object_id:
+        return None
+
+    return meta.Session.query(HDXGeneralToken) \
+        .filter(HDXGeneralToken.token_type == token_type.value) \
+        .filter(HDXGeneralToken.user_id == user_id) \
+        .filter(HDXGeneralToken.object_type == object_type.value) \
+        .filter(HDXGeneralToken.object_id == object_id) \
+        .filter(HDXGeneralToken.state == 'active') \
+        .first()
+
+
 def validate_token(session: AlchemySession, token: str, token_type: TokenType, inactivate=True) -> HDXGeneralToken:
     token_obj = get_by_token(token)
     if not token_obj:

ckanext-hdx_users/ckanext/hdx_users/views/notification_platform.py

@@ -41,11 +41,15 @@ def subscribe_to_dataset() -> Response:
         return tk.redirect_to(dataset_list_url)
 
     context = {'ignore_auth': True}
-    data_dict = {'email': email, 'dataset_id': dataset_id}
 
     try:
+        unsubscribe_token = notification_platform_logic.get_or_generate_unsubscribe_token(email, dataset_id)
+        data_dict = {
+            'email': email,
+            'dataset_id': dataset_id,
+            'unsubscribe_token': unsubscribe_token.token,
+        }
         result = tk.get_action('hdx_add_notification_subscription')(context, data_dict)
-        notification_platform_logic.generate_unsubscribe_token(email, dataset_id)
         _h.flash_success(tk._(
             u'You have successfully set up email notifications for this dataset. These will be sent to {0} when the '
             u'dataset is updated on HDX.'.format(
@@ -73,12 +77,12 @@ def subscription_confirmation() -> Response:
             raise tk.Invalid(tk._('Email address is missing'))
         hdx_validate_email(email)
 
-        token_obj = notification_platform_logic.generate_email_validation_token(email, dataset_id)
+        token_obj = notification_platform_logic.get_or_generate_email_validation_token(email, dataset_id)
 
         subject = u'Please verify your email address'
         verify_email_link = _h.url_for(
             'hdx_notifications.subscribe_to_dataset',
-            email=email, dataset_id=dataset_id, token=token_obj.token, qualified=True
+            token=token_obj.token, qualified=True
         )
         email_data = {
             'verify_email_link': verify_email_link