Update rest end point to wrote to the DB not the queue.

We wrote to the queue as a performance optimisation , however dev found it confusing to not have immediate access to the notification as it may be perished some minutes later under periods of load. Additionally we had a couple of DB issues which led to us dropping notifications.

Pushing the DB write to earlier in the flow makes the system a little  more robust in the early days, we may want to change this when the traffic increases.

app/notifications/rest.py

@@ -14,8 +14,9 @@ from notifications_utils.template import Template
 from notifications_utils.renderers import PassThrough
 from app.clients.email.aws_ses import get_aws_responses
 from app import api_user, encryption, create_uuid, DATETIME_FORMAT, DATE_FORMAT, statsd_client
+from app.dao.notifications_dao import dao_create_notification
 from app.dao.services_dao import dao_fetch_todays_stats_for_service
-from app.models import KEY_TYPE_TEAM, KEY_TYPE_TEST
+from app.models import KEY_TYPE_TEAM, KEY_TYPE_TEST, Notification, KEY_TYPE_NORMAL, EMAIL_TYPE
 from app.dao import (
     templates_dao,
     services_dao,
@@ -44,8 +45,10 @@ from app.errors import (
     register_errors,
     InvalidRequest
 )
+from app.celery.provider_tasks import send_sms_to_provider, send_email_to_provider
 
 register_errors(notifications)
+from app.celery.provider_tasks import send_sms_to_provider, send_email_to_provider
 
 
 @notifications.route('/notifications/email/ses', methods=['POST'])
@@ -194,10 +197,7 @@ def send_notification(notification_type):
 
     service_stats = sum(row.count for row in dao_fetch_todays_stats_for_service(service.id))
 
-    if all((
-        api_user.key_type != KEY_TYPE_TEST,
-        service_stats >= service.message_limit
-    )):
+    if all((api_user.key_type != KEY_TYPE_TEST, service_stats >= service.message_limit)):
         error = 'Exceeded send limits ({}) for today'.format(service.message_limit)
         raise InvalidRequest(error, status_code=429)
 
@@ -242,14 +242,14 @@ def send_notification(notification_type):
         raise InvalidRequest(errors, status_code=400)
 
     if all((
-        api_user.key_type != KEY_TYPE_TEST,
-        service.restricted or api_user.key_type == KEY_TYPE_TEAM,
-        not allowed_to_send_to(
-            notification['to'],
-            itertools.chain.from_iterable(
-                [user.mobile_number, user.email_address] for user in service.users
+            api_user.key_type != KEY_TYPE_TEST,
+            service.restricted or api_user.key_type == KEY_TYPE_TEAM,
+            not allowed_to_send_to(
+                notification['to'],
+                itertools.chain.from_iterable(
+                    [user.mobile_number, user.email_address] for user in service.users
+                )
             )
-        )
     )):
         if (api_user.key_type == KEY_TYPE_TEAM):
             message = 'Can’t send to this recipient using a team-only API key'
@@ -265,34 +265,15 @@ def send_notification(notification_type):
 
     notification_id = create_uuid()
     notification.update({"template_version": template.version})
-    if notification_type == SMS_TYPE:
-        send_sms.apply_async(
-            (
-                service_id,
-                notification_id,
-                encryption.encrypt(notification),
-                datetime.utcnow().strftime(DATETIME_FORMAT)
-            ),
-            kwargs={
-                'api_key_id': str(api_user.id),
-                'key_type': api_user.key_type
-            },
-            queue='db-sms'
-        )
-    else:
-        send_email.apply_async(
-            (
-                service_id,
-                notification_id,
-                encryption.encrypt(notification),
-                datetime.utcnow().strftime(DATETIME_FORMAT)
-            ),
-            kwargs={
-                'api_key_id': str(api_user.id),
-                'key_type': api_user.key_type
-            },
-            queue='db-email'
-        )
+    persist_notification(
+        service,
+        notification_id,
+        notification,
+        datetime.utcnow().strftime(DATETIME_FORMAT),
+        notification_type,
+        str(api_user.id),
+        api_user.key_type
+    )
 
     return jsonify(
         data=get_notification_return_data(
@@ -323,3 +304,28 @@ def get_notification_statistics_for_day():
     )
     data, errors = notifications_statistics_schema.dump(statistics, many=True)
     return jsonify(data=data), 200
+
+
+def persist_notification(
+        service,
+        notification_id,
+        notification,
+        created_at,
+        notification_type,
+        api_key_id=None,
+        key_type=KEY_TYPE_NORMAL,
+):
+    dao_create_notification(
+        Notification.from_api_request(
+            created_at, notification, notification_id, service.id, notification_type, api_key_id, key_type
+        )
+    )
+
+    if notification_type == SMS_TYPE:
+        send_sms_to_provider.apply_async((str(service.id), str(notification_id)), queue='send-sms')
+    if notification_type == EMAIL_TYPE:
+        send_email_to_provider.apply_async((str(service.id), str(notification_id)), queue='send-email')
+
+    current_app.logger.info(
+        "{} {} created at {}".format(notification_type, notification_id, created_at)
+    )