Merge pull request #2771 from alphagov/take-pressure-off-api-db

Reduce the pressure on the db for API post email requests.

app/celery/tasks.py

@@ -18,7 +18,7 @@ from requests import (
     request,
     RequestException
 )
-from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.exc import SQLAlchemyError, IntegrityError
 
 from app import (
     create_uuid,
@@ -293,6 +293,52 @@ def save_email(self,
         handle_exception(self, notification, notification_id, e)
 
 
+@notify_celery.task(bind=True, name="save-api-email", max_retries=5, default_retry_delay=300)
+@statsd(namespace="tasks")
+def save_api_email(self,
+                   encrypted_notification,
+                   ):
+
+    notification = encryption.decrypt(encrypted_notification)
+    service = dao_fetch_service_by_id(notification['service_id'])
+
+    try:
+        current_app.logger.info(f"Persisting notification {notification['id']}")
+
+        persist_notification(
+            notification_id=notification["id"],
+            template_id=notification['template_id'],
+            template_version=notification['template_version'],
+            recipient=notification['to'],
+            service=service,
+            personalisation=notification.get('personalisation'),
+            notification_type=EMAIL_TYPE,
+            client_reference=notification['client_reference'],
+            api_key_id=notification.get('api_key_id'),
+            key_type=KEY_TYPE_NORMAL,
+            created_at=notification['created_at'],
+            reply_to_text=notification['reply_to_text'],
+            status=notification['status'],
+            document_download_count=notification['document_download_count']
+        )
+
+        q = QueueNames.SEND_EMAIL if not service.research_mode else QueueNames.RESEARCH_MODE
+        provider_tasks.deliver_email.apply_async(
+            [notification['id']],
+            queue=q
+        )
+        current_app.logger.info(f"Email {notification['id']} has been persisted.")
+    except IntegrityError:
+        current_app.logger.info(f"Email {notification['id']} already exists.")
+
+    except SQLAlchemyError:
+
+        try:
+            self.retry(queue=QueueNames.RETRY)
+        except self.MaxRetriesExceededError:
+            current_app.logger.error('Max retry failed' + f"Failed to persist notification {notification['id']}")
+
+
 @notify_celery.task(bind=True, name="save-letter", max_retries=5, default_retry_delay=300)
 @statsd(namespace="tasks")
 def save_letter(

app/config.py

@@ -31,6 +31,7 @@ class QueueNames(object):
     SMS_CALLBACKS = 'sms-callbacks'
     ANTIVIRUS = 'antivirus-tasks'
     SANITISE_LETTERS = 'sanitise-letter-tasks'
+    SAVE_API_EMAIL = 'save-api-email-tasks'
 
     @staticmethod
     def all_queues():
@@ -49,6 +50,7 @@ class QueueNames(object):
             QueueNames.CALLBACKS,
             QueueNames.LETTERS,
             QueueNames.SMS_CALLBACKS,
+            QueueNames.SAVE_API_EMAIL
         ]
 
 
@@ -161,6 +163,8 @@ class Config(object):
     MOU_SIGNED_ON_BEHALF_ON_BEHALF_RECEIPT_TEMPLATE_ID = '522b6657-5ca5-4368-a294-6b527703bd0b'
     MOU_NOTIFY_TEAM_ALERT_TEMPLATE_ID = 'd0e66c4c-0c50-43f0-94f5-f85b613202d4'
 
+    HIGH_VOLUME_SERVICE = '539d63a1-701d-400d-ab11-f3ee2319d4d4'
+
     BROKER_URL = 'sqs://'
     BROKER_TRANSPORT_OPTIONS = {
         'region': AWS_REGION,

app/v2/notifications/post_notifications.py

@@ -1,12 +1,22 @@
 import base64
 import functools
+import uuid
+from datetime import datetime
 
 from flask import request, jsonify, current_app, abort
 from notifications_utils.recipients import try_validate_and_format_phone_number
 
-from app import api_user, authenticated_service, notify_celery, document_download_client
+from app import (
+    api_user,
+    authenticated_service,
+    notify_celery,
+    document_download_client,
+    encryption,
+    DATETIME_FORMAT
+)
 from app.celery.letters_pdf_tasks import create_letters_pdf, process_virus_scan_passed
 from app.celery.research_mode_tasks import create_fake_letter_response_file
+from app.celery.tasks import save_api_email
 from app.clients.document_download import DocumentDownloadError
 from app.config import QueueNames, TaskNames
 from app.dao.notifications_dao import update_notification_status_by_reference
@@ -17,13 +27,14 @@ from app.models import (
     EMAIL_TYPE,
     LETTER_TYPE,
     PRIORITY,
+    KEY_TYPE_NORMAL,
     KEY_TYPE_TEST,
     KEY_TYPE_TEAM,
     NOTIFICATION_CREATED,
     NOTIFICATION_SENDING,
     NOTIFICATION_DELIVERED,
     NOTIFICATION_PENDING_VIRUS_CHECK,
-)
+    Notification)
 from app.notifications.process_letter_notifications import (
     create_letter_notification
 )
@@ -192,6 +203,24 @@ def process_sms_or_email_notification(*, form, notification_type, api_key, templ
         simulated=simulated
     )
 
+    if str(service.id) == current_app.config.get('HIGH_VOLUME_SERVICE') and api_key.key_type == KEY_TYPE_NORMAL \
+       and notification_type == EMAIL_TYPE:
+        # Put GOV.UK Email notifications onto a queue
+        # To take the pressure off the db for API requests put the notification for our high volume service onto a queue
+        # the task will then save the notification, then call send_notification_to_queue.
+        # We know that this team does not use the GET request, but relies on callbacks to get the status updates.
+        notification = save_email_to_queue(
+            form=form,
+            notification_type=notification_type,
+            api_key=api_key,
+            template=template,
+            service_id=service.id,
+            personalisation=personalisation,
+            document_download_count=document_download_count,
+            reply_to_text=reply_to_text
+        )
+        return notification
+
     notification = persist_notification(
         template_id=template.id,
         template_version=template.version,
@@ -224,6 +253,41 @@ def process_sms_or_email_notification(*, form, notification_type, api_key, templ
     return notification
 
 
+def save_email_to_queue(
+    *,
+    form,
+    notification_type,
+    api_key,
+    template,
+    service_id,
+    personalisation,
+    document_download_count,
+    reply_to_text=None
+):
+    data = {
+        "id": str(uuid.uuid4()),
+        "template_id": str(template.id),
+        "template_version": template.version,
+        "to": form['email_address'],
+        "service_id": str(service_id),
+        "personalisation": personalisation,
+        "notification_type": notification_type,
+        "api_key_id": str(api_key.id),
+        "key_type": api_key.key_type,
+        "client_reference": form.get('reference', None),
+        "reply_to_text": reply_to_text,
+        "document_download_count": document_download_count,
+        "status": NOTIFICATION_CREATED,
+        "created_at": datetime.utcnow().strftime(DATETIME_FORMAT),
+    }
+    encrypted = encryption.encrypt(
+        data
+    )
+
+    save_api_email.apply_async([encrypted], queue=QueueNames.SAVE_API_EMAIL)
+    return Notification(**data)
+
+
 def process_document_uploads(personalisation_data, service, simulated=False):
     """
     Returns modified personalisation dict and a count of document uploads. If there are no document uploads, returns

manifest.yml.j2

@@ -37,6 +37,7 @@
   'notify-delivery-worker-internal': {},
   'notify-delivery-worker-receipts': {},
   'notify-delivery-worker-service-callbacks': {'disk_quota': '2G'},
+  'notify-delivery-worker-save-api-notifications': {'disk_quota': '2G'},
 } -%}
 
 {%- set app = app_vars[CF_APP] -%}

scripts/paas_app_wrapper.sh

@@ -49,6 +49,10 @@ case $NOTIFY_APP_NAME in
     exec scripts/run_app_paas.sh celery -A run_celery.notify_celery worker --loglevel=INFO --concurrency=11 \
     -Q service-callbacks 2> /dev/null
     ;;
+  delivery-worker-save-api-notifications)
+    exec scripts/run_app_paas.sh celery -A run_celery.notify_celery worker --loglevel=INFO --concurrency=11 \
+    -Q save-api-email-tasks 2> /dev/null
+    ;;
   delivery-celery-beat)
     exec scripts/run_app_paas.sh celery -A run_celery.notify_celery beat --loglevel=INFO
     ;;

tests/app/celery/test_tasks.py

@@ -30,7 +30,7 @@ from app.celery.tasks import (
     s3,
     send_inbound_sms_to_service,
     process_returned_letters_list,
-)
+    save_api_email)
 from app.config import QueueNames
 from app.dao import jobs_dao, service_email_reply_to_dao, service_sms_sender_dao
 from app.models import (
@@ -44,8 +44,8 @@ from app.models import (
     JOB_STATUS_IN_PROGRESS,
     LETTER_TYPE,
     SMS_TYPE,
-    ReturnedLetter
-)
+    ReturnedLetter,
+    NOTIFICATION_CREATED)
 
 from tests.app import load_example_csv
 
@@ -60,8 +60,8 @@ from tests.app.db import (
     create_user,
     create_reply_to_email,
     create_service_with_defined_sms_sender,
-    create_notification_history
-)
+    create_notification_history,
+    create_api_key)
 from tests.conftest import set_config_values
 
 
@@ -1664,3 +1664,75 @@ def test_process_returned_letters_populates_returned_letters_table(
 
     returned_letters = ReturnedLetter.query.all()
     assert len(returned_letters) == 2
+
+
+@freeze_time('2020-03-25 14:30')
+def test_save_api_email(sample_email_template, mocker):
+    mock_send_email_to_provider = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
+    api_key = create_api_key(service=sample_email_template.service)
+    data = {
+        "id": str(uuid.uuid4()),
+        "template_id": str(sample_email_template.id),
+        "template_version": sample_email_template.version,
+        "to": "jane.citizen@example.com",
+        "service_id": str(sample_email_template.service_id),
+        "personalisation": None,
+        "notification_type": sample_email_template.template_type,
+        "api_key_id": str(api_key.id),
+        "key_type": api_key.key_type,
+        "client_reference": 'our email',
+        "reply_to_text": "our.email@gov.uk",
+        "document_download_count": 0,
+        "status": NOTIFICATION_CREATED,
+        "created_at": datetime.utcnow().strftime(DATETIME_FORMAT),
+    }
+
+    encrypted = encryption.encrypt(
+        data
+    )
+
+    assert len(Notification.query.all()) == 0
+    save_api_email(encrypted)
+    notifications = Notification.query.all()
+    assert len(notifications) == 1
+    assert str(notifications[0].id) == data['id']
+    assert notifications[0].created_at == datetime(2020, 3, 25, 14, 30)
+    mock_send_email_to_provider.assert_called_once_with([data['id']], queue=QueueNames.SEND_EMAIL)
+
+
+@freeze_time('2020-03-25 14:30')
+def test_save_api_email_dont_retry_if_notification_already_exists(sample_email_template, mocker):
+    mock_send_email_to_provider = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
+    api_key = create_api_key(service=sample_email_template.service)
+    data = {
+        "id": str(uuid.uuid4()),
+        "template_id": str(sample_email_template.id),
+        "template_version": sample_email_template.version,
+        "to": "jane.citizen@example.com",
+        "service_id": str(sample_email_template.service_id),
+        "personalisation": None,
+        "notification_type": sample_email_template.template_type,
+        "api_key_id": str(api_key.id),
+        "key_type": api_key.key_type,
+        "client_reference": 'our email',
+        "reply_to_text": "our.email@gov.uk",
+        "document_download_count": 0,
+        "status": NOTIFICATION_CREATED,
+        "created_at": datetime.utcnow().strftime(DATETIME_FORMAT),
+    }
+
+    encrypted = encryption.encrypt(
+        data
+    )
+    assert len(Notification.query.all()) == 0
+    save_api_email(encrypted)
+    notifications = Notification.query.all()
+    assert len(notifications) == 1
+    # call the task again with the same notification
+    save_api_email(encrypted)
+    notifications = Notification.query.all()
+    assert len(notifications) == 1
+    assert str(notifications[0].id) == data['id']
+    assert notifications[0].created_at == datetime(2020, 3, 25, 14, 30)
+    # should only have sent the notification once.
+    mock_send_email_to_provider.assert_called_once_with([data['id']], queue=QueueNames.SEND_EMAIL)

tests/app/conftest.py

@@ -491,11 +491,6 @@ def sample_notification_history(notify_db, notify_db_session, sample_template):
     return notification_history
 
 
-@pytest.fixture(scope='function')
-def mock_encryption(mocker):
-    return mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
-
-
 @pytest.fixture(scope='function')
 def sample_invited_user(notify_db_session):
     service = create_service(check_if_service_exists=True)

tests/app/test_config.py

@@ -65,7 +65,7 @@ def test_cloudfoundry_config_has_different_defaults():
 def test_queue_names_all_queues_correct():
     # Need to ensure that all_queues() only returns queue names used in API
     queues = QueueNames.all_queues()
-    assert len(queues) == 14
+    assert len(queues) == 15
     assert set([
         QueueNames.PRIORITY,
         QueueNames.PERIODIC,
@@ -81,4 +81,5 @@ def test_queue_names_all_queues_correct():
         QueueNames.CALLBACKS,
         QueueNames.LETTERS,
         QueueNames.SMS_CALLBACKS,
+        QueueNames.SAVE_API_EMAIL
     ]) == set(queues)

tests/app/v2/notifications/test_post_notifications.py

@@ -1,4 +1,5 @@
 import uuid
+from unittest import mock
 from unittest.mock import call
 
 import pytest
@@ -936,8 +937,7 @@ def test_post_notification_returns_201_when_content_type_is_missing_but_payload_
 
 
 @pytest.mark.parametrize('notification_type', ['email', 'sms'])
-def test_post_email_notification_when_data_is_empty_returns_400(
-        client, sample_service, notification_type):
+def test_post_email_notification_when_data_is_empty_returns_400(client, sample_service, notification_type):
     auth_header = create_authorization_header(service_id=sample_service.id)
     data = None
     response = client.post(
@@ -951,3 +951,86 @@ def test_post_email_notification_when_data_is_empty_returns_400(
         assert error_msg == 'phone_number is a required property'
     else:
         assert error_msg == 'email_address is a required property'
+
+
+def test_post_notifications_saves_email_to_queue(client, notify_db_session, mocker):
+    save_email_task = mocker.patch("app.celery.tasks.save_api_email.apply_async")
+
+    service = create_service(service_id='539d63a1-701d-400d-ab11-f3ee2319d4d4', service_name='high volume service')
+    template = create_template(service=service, content='((message))', template_type=EMAIL_TYPE)
+    data = {
+        "email_address": "joe.citizen@example.com",
+        "template_id": template.id,
+        "personalisation": {"message": "Dear citizen, have a nice day"}
+    }
+    response = client.post(
+        path='/v2/notifications/email',
+        data=json.dumps(data),
+        headers=[('Content-Type', 'application/json'), create_authorization_header(service_id=service.id)]
+    )
+
+    json_resp = response.get_json()
+
+    assert response.status_code == 201
+    assert json_resp['id']
+    assert json_resp['content']['body'] == "Dear citizen, have a nice day"
+    assert json_resp['template']['id'] == str(template.id)
+    save_email_task.assert_called_once_with([mock.ANY], queue='save-api-email-tasks')
+    assert len(Notification.query.all()) == 0
+
+
+def test_post_notifications_doesnt_save_email_to_queue_for_test_emails(client, notify_db_session, mocker):
+    save_email_task = mocker.patch("app.celery.tasks.save_api_email.apply_async")
+    mocked_send_task = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
+
+    service = create_service(service_id='539d63a1-701d-400d-ab11-f3ee2319d4d4', service_name='high volume service')
+    # create_api_key(service=service, key_type='test')
+    template = create_template(service=service, content='((message))', template_type=EMAIL_TYPE)
+    data = {
+        "email_address": "joe.citizen@example.com",
+        "template_id": template.id,
+        "personalisation": {"message": "Dear citizen, have a nice day"}
+    }
+    response = client.post(
+        path='/v2/notifications/email',
+        data=json.dumps(data),
+        headers=[('Content-Type', 'application/json'),
+                 create_authorization_header(service_id=service.id, key_type='test')]
+    )
+
+    json_resp = response.get_json()
+
+    assert response.status_code == 201
+    assert json_resp['id']
+    assert json_resp['content']['body'] == "Dear citizen, have a nice day"
+    assert json_resp['template']['id'] == str(template.id)
+    assert mocked_send_task.called
+    assert not save_email_task.called
+    assert len(Notification.query.all()) == 1
+
+
+def test_post_notifications_doesnt_save_email_to_queue_for_sms(client, notify_db_session, mocker):
+    save_email_task = mocker.patch("app.celery.tasks.save_api_email.apply_async")
+    mocked_send_task = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
+
+    service = create_service(service_id='539d63a1-701d-400d-ab11-f3ee2319d4d4', service_name='high volume service')
+    template = create_template(service=service, content='((message))', template_type=SMS_TYPE)
+    data = {
+        "phone_number": '+447700900855',
+        "template_id": template.id,
+        "personalisation": {"message": "Dear citizen, have a nice day"}
+    }
+    response = client.post(
+        path='/v2/notifications/sms',
+        data=json.dumps(data),
+        headers=[('Content-Type', 'application/json'), create_authorization_header(service_id=service.id)]
+    )
+
+    json_resp = response.get_json()
+
+    assert response.status_code == 201
+    assert json_resp['id']
+    assert mocked_send_task.called
+    assert not save_email_task.called
+
+    assert len(Notification.query.all()) == 1