Merge pull request #1280 from alphagov/letter-api-not-by-job

Letter api not by job

app/aws/s3.py

@@ -72,7 +72,7 @@ def remove_s3_object(bucket_name, object_key):
 
 
 def remove_transformed_dvla_file(job_id):
-    bucket_name = current_app.config['DVLA_UPLOAD_BUCKET_NAME']
+    bucket_name = current_app.config['DVLA_BUCKETS']['job']
     file_location = '{}-dvla-job.text'.format(job_id)
     obj = get_s3_object(bucket_name, file_location)
     return obj.delete()

app/celery/scheduled_tasks.py

@@ -5,6 +5,7 @@ from datetime import (
 
 from flask import current_app
 from sqlalchemy.exc import SQLAlchemyError
+from notifications_utils.s3 import s3upload
 
 from app.aws import s3
 from app import notify_celery
@@ -27,7 +28,9 @@ from app.dao.notifications_dao import (
     is_delivery_slow_for_provider,
     delete_notifications_created_more_than_a_week_ago_by_type,
     dao_get_scheduled_notifications,
-    set_scheduled_notification_to_processed)
+    set_scheduled_notification_to_processed,
+    dao_set_created_live_letter_api_notifications_to_pending,
+)
 from app.dao.statistics_dao import dao_timeout_job_statistics
 from app.dao.provider_details_dao import (
     get_current_provider,
@@ -37,7 +40,7 @@ from app.dao.users_dao import delete_codes_older_created_more_than_a_day_ago
 from app.models import LETTER_TYPE, JOB_STATUS_READY_TO_SEND
 from app.notifications.process_notifications import send_notification_to_queue
 from app.statsd_decorators import statsd
-from app.celery.tasks import process_job
+from app.celery.tasks import process_job, create_dvla_file_contents_for_notifications
 from app.config import QueueNames, TaskNames
 from app.utils import convert_utc_to_bst
 
@@ -313,9 +316,41 @@ def populate_monthly_billing():
 @statsd(namespace="tasks")
 def run_letter_jobs():
     job_ids = dao_get_letter_job_ids_by_status(JOB_STATUS_READY_TO_SEND)
-    notify_celery.send_task(
-        name=TaskNames.DVLA_JOBS,
-        args=(job_ids,),
-        queue=QueueNames.PROCESS_FTP
-    )
-    current_app.logger.info("Queued {} ready letter job ids onto {}".format(len(job_ids), QueueNames.PROCESS_FTP))
+    if job_ids:
+        notify_celery.send_task(
+            name=TaskNames.DVLA_JOBS,
+            args=(job_ids,),
+            queue=QueueNames.PROCESS_FTP
+        )
+        current_app.logger.info("Queued {} ready letter job ids onto {}".format(len(job_ids), QueueNames.PROCESS_FTP))
+
+
+@notify_celery.task(name="run-letter-api-notifications")
+@statsd(namespace="tasks")
+def run_letter_api_notifications():
+    current_time = datetime.utcnow().isoformat()
+
+    notifications = dao_set_created_live_letter_api_notifications_to_pending()
+
+    if notifications:
+        file_contents = create_dvla_file_contents_for_notifications(notifications)
+
+        filename = '{}-dvla-notifications.txt'.format(current_time)
+        s3upload(
+            filedata=file_contents + '\n',
+            region=current_app.config['AWS_REGION'],
+            bucket_name=current_app.config['DVLA_BUCKETS']['notification'],
+            file_location=filename
+        )
+
+        notify_celery.send_task(
+            name=TaskNames.DVLA_NOTIFICATIONS,
+            kwargs={'filename': filename},
+            queue=QueueNames.PROCESS_FTP
+        )
+        current_app.logger.info(
+            "Queued {} ready letter api notifications onto {}".format(
+                len(notifications),
+                QueueNames.PROCESS_FTP
+            )
+        )

app/celery/tasks.py

@@ -27,7 +27,11 @@ from app.dao.jobs_dao import (
     all_notifications_are_created_for_job,
     dao_get_all_notifications_for_job,
     dao_update_job_status)
-from app.dao.notifications_dao import get_notification_by_id, dao_update_notifications_sent_to_dvla
+from app.dao.notifications_dao import (
+    get_notification_by_id,
+    dao_update_notifications_for_job_to_sent_to_dvla,
+    dao_update_notifications_by_reference
+)
 from app.dao.provider_details_dao import get_current_provider
 from app.dao.service_inbound_api_dao import get_service_inbound_api_for_service
 from app.dao.services_dao import dao_fetch_service_by_id, fetch_todays_total_message_count
@@ -42,7 +46,10 @@ from app.models import (
     JOB_STATUS_IN_PROGRESS,
     JOB_STATUS_FINISHED,
     JOB_STATUS_READY_TO_SEND,
-    JOB_STATUS_SENT_TO_DVLA, JOB_STATUS_ERROR)
+    JOB_STATUS_SENT_TO_DVLA, JOB_STATUS_ERROR,
+    NOTIFICATION_SENDING,
+    NOTIFICATION_TECHNICAL_FAILURE
+)
 from app.notifications.process_notifications import persist_notification
 from app.service.utils import service_allowed_to_send_to
 from app.statsd_decorators import statsd
@@ -279,11 +286,11 @@ def persist_letter(
 def build_dvla_file(self, job_id):
     try:
         if all_notifications_are_created_for_job(job_id):
-            file_contents = create_dvla_file_contents(job_id)
+            file_contents = create_dvla_file_contents_for_job(job_id)
             s3upload(
                 filedata=file_contents + '\n',
                 region=current_app.config['AWS_REGION'],
-                bucket_name=current_app.config['DVLA_UPLOAD_BUCKET_NAME'],
+                bucket_name=current_app.config['DVLA_BUCKETS']['job'],
                 file_location="{}-dvla-job.text".format(job_id)
             )
             dao_update_job_status(job_id, JOB_STATUS_READY_TO_SEND)
@@ -302,7 +309,7 @@ def update_job_to_sent_to_dvla(self, job_id):
     # and update all notifications for this job to sending, provider = DVLA
     provider = get_current_provider(LETTER_TYPE)
 
-    updated_count = dao_update_notifications_sent_to_dvla(job_id, provider.identifier)
+    updated_count = dao_update_notifications_for_job_to_sent_to_dvla(job_id, provider.identifier)
     dao_update_job_status(job_id, JOB_STATUS_SENT_TO_DVLA)
 
     current_app.logger.info("Updated {} letter notifications to sending. "
@@ -316,7 +323,48 @@ def update_dvla_job_to_error(self, job_id):
     current_app.logger.info("Updated {} job to {}".format(job_id, JOB_STATUS_ERROR))
 
 
-def create_dvla_file_contents(job_id):
+@notify_celery.task(bind=True, name='update-letter-notifications-to-sent')
+@statsd(namespace="tasks")
+def update_letter_notifications_to_sent_to_dvla(self, notification_references):
+    # This task will be called by the FTP app to update notifications as sent to DVLA
+    provider = get_current_provider(LETTER_TYPE)
+
+    updated_count = dao_update_notifications_by_reference(
+        notification_references,
+        {
+            'status': NOTIFICATION_SENDING,
+            'sent_by': provider.identifier,
+            'sent_at': datetime.utcnow(),
+            'updated_at': datetime.utcnow()
+        }
+    )
+
+    current_app.logger.info("Updated {} letter notifications to sending".format(updated_count))
+
+
+@notify_celery.task(bind=True, name='update-letter-notifications-to-error')
+@statsd(namespace="tasks")
+def update_letter_notifications_to_error(self, notification_references):
+    # This task will be called by the FTP app to update notifications as sent to DVLA
+
+    updated_count = dao_update_notifications_by_reference(
+        notification_references,
+        {
+            'status': NOTIFICATION_TECHNICAL_FAILURE,
+            'updated_at': datetime.utcnow()
+        }
+    )
+
+    current_app.logger.info("Updated {} letter notifications to technical-failure".format(updated_count))
+
+
+def create_dvla_file_contents_for_job(job_id):
+    notifications = dao_get_all_notifications_for_job(job_id)
+
+    return create_dvla_file_contents_for_notifications(notifications)
+
+
+def create_dvla_file_contents_for_notifications(notifications):
     file_contents = '\n'.join(
         str(LetterDVLATemplate(
             notification.template.__dict__,
@@ -325,7 +373,7 @@ def create_dvla_file_contents(job_id):
             contact_block=notification.service.letter_contact_block,
             org_id=notification.service.dvla_organisation.id,
         ))
-        for notification in dao_get_all_notifications_for_job(job_id)
+        for notification in notifications
     )
     return file_contents
 

app/config.py

@@ -49,7 +49,7 @@ class QueueNames(object):
 
 class TaskNames(object):
     DVLA_JOBS = 'send-jobs-to-dvla'
-    DVLA_NOTIFICATIONS = 'send-notifications-to-dvla'
+    DVLA_NOTIFICATIONS = 'send-api-notifications-to-dvla'
 
 
 class Config(object):
@@ -137,7 +137,7 @@ class Config(object):
         'visibility_timeout': 310,
         'queue_name_prefix': NOTIFICATION_QUEUE_PREFIX
     }
-    CELERY_ENABLE_UTC = True,
+    CELERY_ENABLE_UTC = True
     CELERY_TIMEZONE = 'Europe/London'
     CELERY_ACCEPT_CONTENT = ['json']
     CELERY_TASK_SERIALIZER = 'json'
@@ -165,27 +165,27 @@ class Config(object):
         },
         'delete-sms-notifications': {
             'task': 'delete-sms-notifications',
-            'schedule': crontab(minute=0, hour=0),
+            'schedule': crontab(hour=0, minute=0),
             'options': {'queue': QueueNames.PERIODIC}
         },
         'delete-email-notifications': {
             'task': 'delete-email-notifications',
-            'schedule': crontab(minute=20, hour=0),
+            'schedule': crontab(hour=0, minute=20),
             'options': {'queue': QueueNames.PERIODIC}
         },
         'delete-letter-notifications': {
             'task': 'delete-letter-notifications',
-            'schedule': crontab(minute=40, hour=0),
+            'schedule': crontab(hour=0, minute=40),
             'options': {'queue': QueueNames.PERIODIC}
         },
         'delete-inbound-sms': {
             'task': 'delete-inbound-sms',
-            'schedule': crontab(minute=0, hour=1),
+            'schedule': crontab(hour=1, minute=0),
             'options': {'queue': QueueNames.PERIODIC}
         },
         'send-daily-performance-platform-stats': {
             'task': 'send-daily-performance-platform-stats',
-            'schedule': crontab(minute=0, hour=2),
+            'schedule': crontab(hour=2, minute=0),
             'options': {'queue': QueueNames.PERIODIC}
         },
         'switch-current-sms-provider-on-slow-delivery': {
@@ -195,39 +195,44 @@ class Config(object):
         },
         'timeout-sending-notifications': {
             'task': 'timeout-sending-notifications',
-            'schedule': crontab(minute=0, hour=3),
+            'schedule': crontab(hour=3, minute=0),
             'options': {'queue': QueueNames.PERIODIC}
         },
         'remove_sms_email_jobs': {
             'task': 'remove_csv_files',
-            'schedule': crontab(minute=0, hour=4),
+            'schedule': crontab(hour=4, minute=0),
             'options': {'queue': QueueNames.PERIODIC},
             'kwargs': {'job_types': [EMAIL_TYPE, SMS_TYPE]}
         },
         'remove_letter_jobs': {
             'task': 'remove_csv_files',
-            'schedule': crontab(minute=20, hour=4),
+            'schedule': crontab(hour=4, minute=20),
             'options': {'queue': QueueNames.PERIODIC},
             'kwargs': {'job_types': [LETTER_TYPE]}
         },
         'remove_transformed_dvla_files': {
             'task': 'remove_transformed_dvla_files',
-            'schedule': crontab(minute=40, hour=4),
+            'schedule': crontab(hour=4, minute=40),
             'options': {'queue': QueueNames.PERIODIC}
         },
         'timeout-job-statistics': {
             'task': 'timeout-job-statistics',
-            'schedule': crontab(minute=0, hour=5),
+            'schedule': crontab(hour=5, minute=0),
             'options': {'queue': QueueNames.PERIODIC}
         },
         'populate_monthly_billing': {
             'task': 'populate_monthly_billing',
-            'schedule': crontab(minute=10, hour=5),
+            'schedule': crontab(hour=5, minute=10),
             'options': {'queue': QueueNames.PERIODIC}
         },
         'run-letter-jobs': {
             'task': 'run-letter-jobs',
-            'schedule': crontab(hour=17, minute=30),
+            'schedule': crontab(hour=5, minute=30),
+            'options': {'queue': QueueNames.PERIODIC}
+        },
+        'run-letter-api-notifications': {
+            'task': 'run-letter-api-notifications',
+            'schedule': crontab(hour=5, minute=40),
             'options': {'queue': QueueNames.PERIODIC}
         }
     }
@@ -253,7 +258,10 @@ class Config(object):
     FUNCTIONAL_TEST_PROVIDER_SERVICE_ID = None
     FUNCTIONAL_TEST_PROVIDER_SMS_TEMPLATE_ID = None
 
-    DVLA_UPLOAD_BUCKET_NAME = "{}-dvla-file-per-job".format(os.getenv('NOTIFY_ENVIRONMENT'))
+    DVLA_BUCKETS = {
+        'job': '{}-dvla-file-per-job'.format(os.getenv('NOTIFY_ENVIRONMENT')),
+        'notification': '{}-dvla-letter-api-files'.format(os.getenv('NOTIFY_ENVIRONMENT'))
+    }
 
     API_KEY_LIMITS = {
         KEY_TYPE_TEAM: {

app/dao/notifications_dao.py

@@ -460,7 +460,7 @@ def is_delivery_slow_for_provider(
 
 @statsd(namespace="dao")
 @transactional
-def dao_update_notifications_sent_to_dvla(job_id, provider):
+def dao_update_notifications_for_job_to_sent_to_dvla(job_id, provider):
     now = datetime.utcnow()
     updated_count = db.session.query(
         Notification).filter(Notification.job_id == job_id).update(
@@ -473,6 +473,27 @@ def dao_update_notifications_sent_to_dvla(job_id, provider):
     return updated_count
 
 
+@statsd(namespace="dao")
+@transactional
+def dao_update_notifications_by_reference(references, update_dict):
+    now = datetime.utcnow()
+    updated_count = Notification.query.filter(
+        Notification.reference.in_(references)
+    ).update(
+        update_dict,
+        synchronize_session=False
+    )
+
+    NotificationHistory.query.filter(
+        NotificationHistory.reference.in_(references)
+    ).update(
+        update_dict,
+        synchronize_session=False
+    )
+
+    return updated_count
+
+
 @statsd(namespace="dao")
 def dao_get_notifications_by_to_field(service_id, search_term, statuses=None):
     try:
@@ -555,3 +576,30 @@ def dao_get_total_notifications_sent_per_day_for_performance_platform(start_date
         NotificationHistory.key_type != KEY_TYPE_TEST,
         NotificationHistory.notification_type != LETTER_TYPE
     ).one()
+
+
+def dao_set_created_live_letter_api_notifications_to_pending():
+    """
+    Sets all past scheduled jobs to pending, and then returns them for further processing.
+
+    this is used in the run_scheduled_jobs task, so we put a FOR UPDATE lock on the job table for the duration of
+    the transaction so that if the task is run more than once concurrently, one task will block the other select
+    from completing until it commits.
+    """
+    notifications = db.session.query(
+        Notification
+    ).filter(
+        Notification.notification_type == LETTER_TYPE,
+        Notification.status == NOTIFICATION_CREATED,
+        Notification.key_type == KEY_TYPE_NORMAL,
+        Notification.api_key != None  # noqa
+    ).with_for_update(
+    ).all()
+
+    for notification in notifications:
+        notification.status = NOTIFICATION_PENDING
+
+    db.session.add_all(notifications)
+    db.session.commit()
+
+    return notifications

app/notifications/process_letter_notifications.py

@@ -1,45 +1,23 @@
 from app import create_random_identifier
-from app.models import LETTER_TYPE, JOB_STATUS_READY_TO_SEND, Job
-from app.dao.jobs_dao import dao_create_job
+from app.models import LETTER_TYPE
 from app.notifications.process_notifications import persist_notification
-from app.v2.errors import InvalidRequest
-from app.variables import LETTER_API_FILENAME
 
 
-def create_letter_api_job(template):
-    service = template.service
-    if not service.active:
-        raise InvalidRequest('Service {} is inactive'.format(service.id), 403)
-    if template.archived:
-        raise InvalidRequest('Template {} is deleted'.format(template.id), 400)
-
-    job = Job(
-        original_file_name=LETTER_API_FILENAME,
-        service=service,
-        template=template,
-        template_version=template.version,
-        notification_count=1,
-        job_status=JOB_STATUS_READY_TO_SEND,
-        created_by=None
-    )
-    dao_create_job(job)
-    return job
-
-
-def create_letter_notification(letter_data, job, api_key):
+def create_letter_notification(letter_data, template, api_key, status):
     notification = persist_notification(
-        template_id=job.template.id,
-        template_version=job.template.version,
+        template_id=template.id,
+        template_version=template.version,
         # we only accept addresses_with_underscores from the API (from CSV we also accept dashes, spaces etc)
         recipient=letter_data['personalisation']['address_line_1'],
-        service=job.service,
+        service=template.service,
         personalisation=letter_data['personalisation'],
         notification_type=LETTER_TYPE,
         api_key_id=api_key.id,
         key_type=api_key.key_type,
-        job_id=job.id,
-        job_row_number=0,
+        job_id=None,
+        job_row_number=None,
         reference=create_random_identifier(),
-        client_reference=letter_data.get('reference')
+        client_reference=letter_data.get('reference'),
+        status=status
     )
     return notification

app/notifications/process_notifications.py

@@ -3,6 +3,7 @@ from datetime import datetime
 
 from flask import current_app
 
+from notifications_utils.clients import redis
 from notifications_utils.recipients import (
     get_international_phone_info,
     validate_and_format_phone_number,
@@ -11,10 +12,8 @@ from notifications_utils.recipients import (
 
 from app import redis_store
 from app.celery import provider_tasks
-from notifications_utils.clients import redis
-
 from app.config import QueueNames
-from app.models import SMS_TYPE, Notification, KEY_TYPE_TEST, EMAIL_TYPE, ScheduledNotification
+from app.models import SMS_TYPE, Notification, KEY_TYPE_TEST, EMAIL_TYPE, NOTIFICATION_CREATED, ScheduledNotification
 from app.dao.notifications_dao import (dao_create_notification,
                                        dao_delete_notifications_and_history_by_id,
                                        dao_created_scheduled_notification)
@@ -52,7 +51,8 @@ def persist_notification(
     client_reference=None,
     notification_id=None,
     simulated=False,
-    created_by_id=None
+    created_by_id=None,
+    status=NOTIFICATION_CREATED
 ):
     notification_created_at = created_at or datetime.utcnow()
     if not notification_id:
@@ -73,7 +73,8 @@ def persist_notification(
         job_row_number=job_row_number,
         client_reference=client_reference,
         reference=reference,
-        created_by_id=created_by_id
+        created_by_id=created_by_id,
+        status=status
     )
 
     if notification_type == SMS_TYPE:

app/v2/notifications/post_notifications.py

@@ -4,16 +4,23 @@ from flask import request, jsonify, current_app, abort
 
 from app import api_user, authenticated_service
 from app.config import QueueNames
-from app.dao.jobs_dao import dao_update_job
-from app.models import SMS_TYPE, EMAIL_TYPE, LETTER_TYPE, PRIORITY, KEY_TYPE_TEST, KEY_TYPE_TEAM
-from app.celery.tasks import build_dvla_file, update_job_to_sent_to_dvla
+from app.models import (
+    SMS_TYPE,
+    EMAIL_TYPE,
+    LETTER_TYPE,
+    PRIORITY,
+    KEY_TYPE_TEST,
+    KEY_TYPE_TEAM,
+    NOTIFICATION_CREATED,
+    NOTIFICATION_SENDING
+)
+from app.celery.tasks import update_letter_notifications_to_sent_to_dvla
 from app.notifications.process_notifications import (
     persist_notification,
     send_notification_to_queue,
     simulated_recipient,
     persist_scheduled_notification)
 from app.notifications.process_letter_notifications import (
-    create_letter_api_job,
     create_letter_notification
 )
 from app.notifications.validators import (
@@ -36,7 +43,6 @@ from app.v2.notifications.create_response import (
     create_post_email_response_from_notification,
     create_post_letter_response_from_notification
 )
-from app.variables import LETTER_TEST_API_FILENAME
 
 
 @v2_notification_blueprint.route('/<notification_type>', methods=['POST'])
@@ -153,19 +159,17 @@ def process_letter_notification(*, letter_data, api_key, template):
     if api_key.service.restricted and api_key.key_type != KEY_TYPE_TEST:
         raise BadRequestError(message='Cannot send letters when service is in trial mode', status_code=403)
 
-    job = create_letter_api_job(template)
-    notification = create_letter_notification(letter_data, job, api_key)
+    should_send = not (api_key.service.research_mode or api_key.key_type == KEY_TYPE_TEST)
 
-    if api_key.service.research_mode or api_key.key_type == KEY_TYPE_TEST:
-        # distinguish real API jobs from test jobs by giving the test jobs a different filename
-        job.original_file_name = LETTER_TEST_API_FILENAME
-        dao_update_job(job)
-        update_job_to_sent_to_dvla.apply_async([str(job.id)], queue=QueueNames.RESEARCH_MODE)
-    else:
-        build_dvla_file.apply_async([str(job.id)], queue=QueueNames.JOBS)
+    # if we don't want to actually send the letter, then start it off in SENDING so we don't pick it up
+    status = NOTIFICATION_CREATED if should_send else NOTIFICATION_SENDING
+
+    notification = create_letter_notification(letter_data, template, api_key, status=status)
+
+    if not should_send:
+        update_letter_notifications_to_sent_to_dvla.apply_async(
+            kwargs={'notification_references': [notification.reference]},
+            queue=QueueNames.RESEARCH_MODE
+        )
 
-    current_app.logger.info("send job {} for api notification {} to build-dvla-file in the process-job queue".format(
-        job.id,
-        notification.id
-    ))
     return notification