Optimize bulk SMS processing performance (#2420)

* Optimize bulk SMS processing performance

* Update to latest version of utils

---------

Co-authored-by: Jumana Bahrainwala <[email protected]>
Co-authored-by: Jumana B <[email protected]>

.github/workflows/test.yaml

@@ -69,7 +69,7 @@ jobs:
       run: |
         cp -f .env.example .env
     - name: Checks for new endpoints against AWS WAF rules
-      uses: cds-snc/notification-utils/.github/actions/waffles@53.0.1
+      uses: cds-snc/notification-utils/.github/actions/waffles@53.1.2
       with:
         app-loc: '/github/workspace'
         app-libs: '/github/workspace/env/site-packages'

app/job/rest.py

@@ -31,7 +31,9 @@
     JOB_STATUS_SCHEDULED,
     SMS_TYPE,
 )
-from app.notifications.process_notifications import simulated_recipient
+from app.notifications.process_notifications import (
+    csv_has_simulated_and_non_simulated_recipients,
+)
 from app.notifications.validators import (
     check_email_annual_limit,
     check_email_daily_limit,
@@ -175,16 +177,16 @@ def create_job(service_id):
         data["sender_id"] = data.get("sender_id", default_sender_id)
 
         # calculate the number of simulated recipients
-        numberOfSimulated = sum(simulated_recipient(i["phone_number"].data, template.template_type) for i in recipient_csv.rows)
-        mixedRecipients = numberOfSimulated > 0 and numberOfSimulated != len(recipient_csv)
+        requested_recipients = [i["phone_number"].data for i in list(recipient_csv.get_rows())]
+        has_simulated, has_real_recipients = csv_has_simulated_and_non_simulated_recipients(
+            requested_recipients, template.template_type
+        )
 
-        # if they have specified testing and NON-testing recipients, raise an error
-        if mixedRecipients:
+        if has_simulated and has_real_recipients:
             raise InvalidRequest(message="Bulk sending to testing and non-testing numbers is not supported", status_code=400)
 
-        is_test_notification = len(recipient_csv) == numberOfSimulated
-
-        if not is_test_notification:
+        # Check and track limits if we're not sending test notifications
+        if has_real_recipients and not has_simulated:
             check_sms_annual_limit(service, len(recipient_csv))
             check_sms_daily_limit(service, len(recipient_csv))
             increment_sms_daily_count_send_warnings_if_needed(service, len(recipient_csv))

app/notifications/process_notifications.py

@@ -4,6 +4,7 @@
 
 from flask import current_app
 from notifications_utils.clients import redis
+from notifications_utils.decorators import parallel_process_iterable
 from notifications_utils.recipients import (
     format_email_address,
     get_international_phone_info,
@@ -365,6 +366,54 @@ def persist_notifications(notifications: List[VerifiedNotification]) -> List[Not
     return lofnotifications
 
 
+def csv_has_simulated_and_non_simulated_recipients(
+    to_addresses: set, notification_type: NotificationType, chunk_size=5000
+) -> tuple[bool, bool]:
+    """Kicks off a parallelized process to check if a RecipientCSV contains simulated, non-simulated, or both types of recipients.
+
+    Args:
+        to_addresses (list): A list of recipients pulled from a RecipientCSV.
+        notification_type (NotificationType): The notification type (email | sms)
+
+    Returns:
+        int: The number of recipients that are simulated
+    """
+    simulated_recipients = (
+        current_app.config["SIMULATED_SMS_NUMBERS"]
+        if notification_type == SMS_TYPE
+        else current_app.config["SIMULATED_EMAIL_ADDRESSES"]
+    )
+    results = bulk_simulated_recipients(to_addresses, simulated_recipients)
+
+    found_simulated = any(result[0] for result in results)
+    found_non_simulated = any(result[1] for result in results)
+
+    return found_simulated, found_non_simulated
+
+
+@parallel_process_iterable(break_condition=lambda result: result[0] and result[1])
+def bulk_simulated_recipients(chunk: list, simulated_recipients: tuple):
+    """Parallelized function that processes a chunk of recipients, checking if the chunk contains simulated, non-simulated, or both types of recipients.
+
+    Args:
+        chunk (list): The list of recipients to be processed
+        simulated_recipients (list): The list of simulated recipients from the app's config
+
+    Returns:
+        tuple: Two boolean values indicating if the chunk contains simulated and non-simulated recipients
+    """
+    found_simulated = False
+    found_non_simulated = False
+    for recipient in chunk:
+        if recipient in simulated_recipients:
+            found_simulated = True
+        else:
+            found_non_simulated = True
+        if found_simulated and found_non_simulated:
+            break
+    return found_simulated, found_non_simulated
+
+
 def simulated_recipient(to_address: str, notification_type: NotificationType) -> bool:
     if notification_type == SMS_TYPE:
         formatted_simulated_numbers = [

app/v2/notifications/post_notifications.py

@@ -66,6 +66,7 @@
 from app.notifications.process_letter_notifications import create_letter_notification
 from app.notifications.process_notifications import (
     choose_queue,
+    csv_has_simulated_and_non_simulated_recipients,
     db_save_and_send_notification,
     persist_notification,
     persist_scheduled_notification,
@@ -232,15 +233,6 @@ def post_bulk():
     check_for_csv_errors(recipient_csv, max_rows, remaining_daily_messages, remaining_annual_messages)
     notification_count_requested = len(list(recipient_csv.get_rows()))
 
-    for row in recipient_csv.get_rows():
-        try:
-            validate_template(template.id, row.personalisation, authenticated_service, template.template_type)
-        except BadRequestError as e:
-            message = e.message + ". Notification to {} on row #{} exceeds the maximum size limit.".format(
-                row.recipient, row.index + 1
-            )
-            raise BadRequestError(message=message)
-
     if template.template_type == EMAIL_TYPE and api_user.key_type != KEY_TYPE_TEST:
         check_email_annual_limit(authenticated_service, notification_count_requested)
         check_email_daily_limit(authenticated_service, notification_count_requested)
@@ -257,16 +249,16 @@ def post_bulk():
             form["validated_sender_id"] = default_sender_id
 
         # calculate the number of simulated recipients
-        numberOfSimulated = sum(
-            simulated_recipient(i["phone_number"].data, template.template_type) for i in list(recipient_csv.get_rows())
+        requested_recipients = [i["phone_number"].data for i in list(recipient_csv.get_rows())]
+        has_simulated, has_real_recipients = csv_has_simulated_and_non_simulated_recipients(
+            requested_recipients, template.template_type
         )
-        mixedRecipients = numberOfSimulated > 0 and numberOfSimulated != notification_count_requested
 
         # if its a live or a team key, and they have specified testing and NON-testing recipients, raise an error
-        if api_user.key_type != KEY_TYPE_TEST and mixedRecipients:
+        if api_user.key_type != KEY_TYPE_TEST and (has_simulated and has_real_recipients):
             raise BadRequestError(message="Bulk sending to testing and non-testing numbers is not supported", status_code=400)
 
-        is_test_notification = api_user.key_type == KEY_TYPE_TEST or notification_count_requested == numberOfSimulated
+        is_test_notification = api_user.key_type == KEY_TYPE_TEST or (has_simulated and not has_real_recipients)
 
         if not is_test_notification:
             check_sms_annual_limit(authenticated_service, len(recipient_csv))

pyproject.toml

@@ -65,7 +65,7 @@ Werkzeug = "3.0.4"
 MarkupSafe = "2.1.5"
 # REVIEW: v2 is using sha512 instead of sha1 by default (in v1)
 itsdangerous = "2.2.0"
-notifications-utils = { git = "https://github.com/cds-snc/notifier-utils.git", tag = "53.1.0" }
+notifications-utils = { git = "https://github.com/cds-snc/notifier-utils.git", tag = "53.1.2" }
 
 # rsa = "4.9  # awscli 1.22.38 depends on rsa<4.8
 typing-extensions = "4.12.2"