#2172 - Push Celery Migration

.talismanrc

@@ -15,4 +15,10 @@ fileignoreconfig:
   checksum: 572fb3a52e89ae19c5a94be881bcbfe0247bae112bce58d4ae1e9eff5a5c91b7
 - filename: tests/app/v2/notifications/test_get_notifications.py
   checksum: 167aa80c8aac01566b3cf627cdaa839feac9dc77f0b13b6d0a008035aa8e15f1
+- filename: tests/app/v2/notifications/test_push_broadcast_notifications.py
+  checksum: a2e9c2d625158d39622cfe82f1cc38d9623acc6bf4c2d56ea816910143ad15dd
+- filename: tests/app/v2/notifications/test_push_notifications.py
+  checksum: 8781698e08707a8c9f50fc4f6e4342065a65a069614b88a1192e8ee8f89c268c
+- filename: tests/app/va/vetext/test_client.py
+  checksum: 36fb86b6221e49703e51eda249e1a6d090483968a1c7a198be231470e021642c
 version: "1.0"

app/celery/provider_tasks.py

@@ -1,6 +1,12 @@
 from uuid import UUID
 
-from app import notify_celery
+from celery import Task
+from flask import current_app
+from notifications_utils.field import NullValueForNonConditionalPlaceholderException
+from notifications_utils.recipients import InvalidEmailError, InvalidPhoneError
+from notifications_utils.statsd_decorators import statsd
+
+from app import notify_celery, vetext_client
 from app.celery.common import (
     can_retry,
     handle_max_retries_exceeded,
@@ -28,14 +34,9 @@
     NotificationTechnicalFailureException,
 )
 from app.models import Notification
+from app.v2.dataclasses import V2PushPayload
 from app.v2.errors import RateLimitError
 
-from celery import Task
-from flask import current_app
-from notifications_utils.field import NullValueForNonConditionalPlaceholderException
-from notifications_utils.recipients import InvalidEmailError, InvalidPhoneError
-from notifications_utils.statsd_decorators import statsd
-
 
 # Including sms_sender_id is necessary in case it's passed in when being called
 @notify_celery.task(
@@ -269,3 +270,39 @@ def _handle_delivery_failure(
             if notification:
                 check_and_queue_callback_task(notification)
             raise NotificationTechnicalFailureException(msg)
+
+
+@notify_celery.task(
+    bind=True,
+    serializer='pickle',
+    name='deliver_push',
+    throws=(AutoRetryException,),
+    autoretry_for=(AutoRetryException,),
+    max_retries=2886,
+    retry_backoff=True,
+    retry_backoff_max=60,
+)
+@statsd(namespace='tasks')
+def deliver_push(
+    celery_task: Task,
+    payload: V2PushPayload,
+) -> None:
+    """Deliver a validated push (or broadcast) payload to the provider client."""
+    vetext_payload = vetext_client.format_for_vetext(payload)
+    try:
+        vetext_client.send_push_notification(vetext_payload)
+    except RetryableException:
+        max_retries = celery_task.max_retries
+        retries = celery_task.request.retries
+
+        if retries < max_retries:
+            current_app.logger.warning(
+                'Push notification retrying: %s, max retries: %s, retries: %s', max_retries, retries
+            )
+            raise AutoRetryException('Found RetryableException, autoretrying...')
+        else:
+            current_app.logger.exception('deliver_push: Notification failed by exceeding retry limits')
+            raise NonRetryableException('Exceeded retries')
+    except NonRetryableException:
+        current_app.logger.exception('deliver_push encountered a NonRetryableException')
+        raise

app/config.py

@@ -234,6 +234,7 @@ class Config(object):
             'app.celery.process_pinpoint_receipt_tasks',
             'app.celery.process_pinpoint_inbound_sms',
             'app.celery.process_delivery_status_result_tasks',
+            'app.celery.provider_tasks',
             'app.celery.v3.notification_tasks',
             'app.celery.process_ses_receipts_tasks',
             'app.celery.send_va_profile_notification_status_tasks',
@@ -325,6 +326,7 @@ class Config(object):
         },
         'task_queues': [Queue(queue, Exchange('default'), routing_key=queue) for queue in QueueNames.all_queues()],
         'task_routes': {
+            'app.celery.provider_tasks.deliver_push': {'queue': QueueNames.SEND_EMAIL},
             'app.celery.v3.notification_tasks.v3_process_notification': {'queue': QueueNames.NOTIFY},
             'app.celery.v3.notification_tasks.v3_send_email_notification': {'queue': QueueNames.SEND_EMAIL},
             'app.celery.v3.notification_tasks.v3_send_sms_notification': {'queue': QueueNames.SEND_SMS},

app/internal/rest.py

@@ -14,9 +14,10 @@
     - trace_id
 """
 
+import time
 from werkzeug.exceptions import UnsupportedMediaType
 from contextlib import suppress
-from flask import Blueprint, current_app, request
+from flask import Blueprint, current_app, jsonify, request
 
 
 internal_blueprint = Blueprint('internal', __name__, url_prefix='/internal')
@@ -56,4 +57,29 @@
     else:
         response_body = {generic: request.json}
 
     return response_body, status_code
+
+
+@internal_blueprint.route('/502', methods=['POST', 'GET'])
+def throw_502():
+    """
+    Returns a 502 Bad Gateway response for testing purposes.
+
+    Returns:
+        Response: A JSON response with 'result' set to 'error' and 'message',
+        along with a 502 status code.
+    """
+    return jsonify(result='error', message='hello world'), 502
+
+
+@internal_blueprint.route('/sleep', methods=['POST', 'GET'])
+def sleep():
+    """
+    Simulates a delayed response by sleeping for 30 seconds.
+
+    Returns:
+        Response: A JSON response indicating success, with a 'message' stating the sleep duration,
+          along with a 201 status code.
+    """
+    time.sleep(30)
+    return jsonify(result='success', message='slept for 30s'), 201

app/mobile_app/mobile_app.py

@@ -10,7 +10,7 @@ def __init__(
         self.type: MobileAppType = type
         self.sid = self._get_sid_from_env()
 
-    def _get_sid_from_env(self):
+    def _get_sid_from_env(self) -> str:
         sid = os.getenv(f'{self.type.value}_SID', None)
         if not sid:
             raise ValueError(f'Missing SID for app: {self.type.value}')

app/mobile_app/mobile_app_registry.py

@@ -21,8 +21,12 @@ def __init__(self, logger):
     def get_app(
         self,
         app_type: MobileAppType,
-    ) -> MobileApp | None:
-        return self._registry.get(app_type)
+    ) -> MobileApp:
+        try:
+            return self._registry[app_type]
+        except KeyError:
+            self.logger.warning('Attempted to use an app that is not initialized: %s', app_type)
+            raise
 
     def get_registered_apps(self) -> List[MobileAppType]:
         return list(self._registry.keys())

app/v2/dataclasses.py

@@ -0,0 +1,27 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class V2PushPayload:
+    app_sid: str
+    template_id: str
+    icn: str | None = None
+    topic_sid: str | None = None
+    personalisation: dict[str, str] | None = None
+
+    def is_valid(self) -> bool:
+        """Identify if the payload is valid
+
+        Returns:
+            bool: True if valid
+        """
+        # icn alone means it's push. topic_sid alone means broadcast. Cannot be both.
+        return (self.icn and self.topic_sid is None) or (self.topic_sid and self.icn is None)
+
+    def is_broadcast(self) -> bool:
+        """Identify if the payload is a broadcast
+
+        Returns:
+            bool: True if it is a broadcast
+        """
+        return self.topic_sid is not None

app/v2/notifications/rest_push.py

@@ -1,59 +1,99 @@
-from flask import jsonify, request
+import json
 
-from app import authenticated_service, vetext_client, mobile_app_registry
+from celery.exceptions import CeleryError
+from jsonschema import ValidationError
+from kombu.exceptions import OperationalError
+from flask import current_app, jsonify, request
+
+from app import authenticated_service, mobile_app_registry
+from app.celery.provider_tasks import deliver_push
 from app.constants import PUSH_TYPE
-from app.feature_flags import FeatureFlag, is_feature_enabled
 from app.mobile_app import DEAFULT_MOBILE_APP_TYPE, MobileAppType
 from app.schema_validation import validate
 from app.utils import get_public_notify_type_text
 from app.v2.errors import BadRequestError
 from app.v2.notifications import v2_notification_blueprint
+from app.v2.dataclasses import V2PushPayload
 from app.v2.notifications.notification_schemas import push_notification_broadcast_request, push_notification_request
-from app.va.vetext import VETextBadRequestException, VETextNonRetryableException, VETextRetryableException
 
 
 @v2_notification_blueprint.route('/push', methods=['POST'])
 def send_push_notification():
-    return push_notification_helper(push_notification_request, False)
+    return push_notification_helper(push_notification_request)
 
 
 @v2_notification_blueprint.route('/push/broadcast', methods=['POST'])
 def send_push_broadcast_notification():
-    return push_notification_helper(push_notification_broadcast_request, True)
+    return push_notification_helper(push_notification_broadcast_request)
 
 
-def push_notification_helper(schema: dict, is_broadcast: bool):
+def push_notification_helper(schema: dict):
     """
     Note that this helper cannot be called other than as part of a request because it accesses
     the Flask "request" instance.
     """
-
-    if not is_feature_enabled(FeatureFlag.PUSH_NOTIFICATIONS_ENABLED):
-        raise NotImplementedError()
-
     if not authenticated_service.has_permissions(PUSH_TYPE):
         raise BadRequestError(
-            message='Service is not allowed to send {}'.format(get_public_notify_type_text(PUSH_TYPE, plural=True))
+            message='Service is not allowed to send {}'.format(
+                get_public_notify_type_text(PUSH_TYPE, plural=True),
+            ),
+            status_code=403,
         )
 
-    req_json = validate(request.get_json(), schema)
+    payload = validate_push_payload(schema)
 
-    if 'mobile_app' in req_json:
-        app_instance = mobile_app_registry.get_app(MobileAppType[req_json['mobile_app']])
+    try:
+        # Choosing to use the email queue for push to limit the number of empty queues
+        deliver_push.delay(payload)
+    except (CeleryError, OperationalError):
+        current_app.logger.exception('Failed to enqueue deliver_push request')
+        response = jsonify(result='error', message='VA Notify service impaired, please try again'), 502
     else:
-        app_instance = mobile_app_registry.get_app(DEAFULT_MOBILE_APP_TYPE)
+        response = jsonify(result='success'), 201
+    # Flask turns the tuple into a json and status_code
+    return response
+
+
+def validate_push_payload(schema: dict[str, str]) -> V2PushPayload:
+    """Validate an incoming push request.
+
+    Args:
+        schema (dict[str, str]): The incoming request
 
+    Raises:
+        BadRequestError: Failed validation
+
+    Returns:
+        dict[str, str]: Validated request dictionary
+    """
     try:
-        vetext_client.send_push_notification(
-            app_instance.sid,
-            req_json['template_id'],
-            req_json['topic_sid'] if is_broadcast else req_json['recipient_identifier']['id_value'],
-            is_broadcast,
-            req_json.get('personalisation'),
-        )
-    except VETextBadRequestException as e:
-        raise BadRequestError(message=e.message, status_code=400)
-    except (VETextNonRetryableException, VETextRetryableException):
-        return jsonify(result='error', message='Invalid response from downstream service'), 502
-    else:
-        return jsonify(result='success'), 201
+        req_json: dict[str, str] = validate(request.get_json(), schema)
+
+        # Validate the application they sent us is valid or use the default
+        if 'mobile_app' in req_json:
+            app_sid = mobile_app_registry.get_app(MobileAppType[req_json['mobile_app']]).sid
+        else:
+            app_sid = mobile_app_registry.get_app(DEAFULT_MOBILE_APP_TYPE).sid
+    except (KeyError, TypeError) as e:
+        current_app.logger.warning('Push request failed validation due to mobile app setup: %s', e)
+        raise BadRequestError(message=str(e), status_code=400)
+    except ValidationError as e:
+        current_app.logger.warning('Push request failed validation: %s', e)
+        error_data = json.loads(e.message)
+        error_data['errors'] = error_data['errors'][0]
+        raise e
+    except Exception:
+        msg = 'Unable to process request for push notification - bad request'
+        current_app.logger.exception(msg)
+        raise BadRequestError(message=msg, status_code=400)
+
+    # Use get() on optionals - schema validated it is correct
+    payload = V2PushPayload(
+        app_sid,
+        req_json['template_id'],
+        req_json.get('recipient_identifier', {}).get('id_value'),  # ICN
+        req_json.get('topic_sid'),
+        req_json.get('personalisation'),
+    )
+    current_app.logger.debug('V2PushPayload is: %s', payload)
+    return payload