merge from main

app/celery/provider_tasks.py

@@ -1,107 +1,19 @@
 import json
 import os
-from datetime import timedelta
 
-from botocore.exceptions import ClientError
 from flask import current_app
 from sqlalchemy.orm.exc import NoResultFound
 
-from app import aws_cloudwatch_client, notify_celery, redis_store
+from app import notify_celery, redis_store
 from app.clients.email import EmailClientNonRetryableException
 from app.clients.email.aws_ses import AwsSesClientThrottlingSendRateException
 from app.clients.sms import SmsClientResponseException
 from app.config import Config, QueueNames
 from app.dao import notifications_dao
-from app.dao.notifications_dao import (
+from app.dao.notifications_dao import update_notification_status_by_id
-    sanitize_successful_notification_by_id,
-    update_notification_status_by_id,
-)
 from app.delivery import send_to_providers
 from app.enums import NotificationStatus
 from app.exceptions import NotificationTechnicalFailureException
-from app.utils import utc_now
-
-# This is the amount of time to wait after sending an sms message before we check the aws logs and look for delivery
-# receipts
-DELIVERY_RECEIPT_DELAY_IN_SECONDS = 30
-
-
-@notify_celery.task(
-    bind=True,
-    name="check_sms_delivery_receipt",
-    max_retries=48,
-    default_retry_delay=300,
-)
-def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
-    """
-    This is called after deliver_sms to check the status of the message. This uses the same number of
-    retries and the same delay period as deliver_sms.  In addition, this fires five minutes after
-    deliver_sms initially. So the idea is that most messages will succeed and show up in the logs quickly.
-    Other message will resolve successfully after a retry or to.  A few will fail but it will take up to
-    4 hours to know for sure.  The call to check_sms will raise an exception if neither a success nor a
-    failure appears in the cloudwatch logs, so this should keep retrying until the log appears, or until
-    we run out of retries.
-    """
-    # TODO the localstack cloudwatch doesn't currently have our log groups.  Possibly create them with awslocal?
-    if aws_cloudwatch_client.is_localstack():
-        status = "success"
-        provider_response = "this is a fake successful localstack sms message"
-        carrier = "unknown"
-    else:
-        try:
-            status, provider_response, carrier = aws_cloudwatch_client.check_sms(
-                message_id, notification_id, sent_at
-            )
-        except NotificationTechnicalFailureException as ntfe:
-            provider_response = "Unable to find carrier response -- still looking"
-            status = "pending"
-            carrier = ""
-            update_notification_status_by_id(
-                notification_id,
-                status,
-                carrier=carrier,
-                provider_response=provider_response,
-            )
-            raise self.retry(exc=ntfe)
-        except ClientError as err:
-            # Probably a ThrottlingException but could be something else
-            error_code = err.response["Error"]["Code"]
-            provider_response = (
-                f"{error_code} while checking sms receipt -- still looking"
-            )
-            status = "pending"
-            carrier = ""
-            update_notification_status_by_id(
-                notification_id,
-                status,
-                carrier=carrier,
-                provider_response=provider_response,
-            )
-            raise self.retry(exc=err)
-
-    if status == "success":
-        status = NotificationStatus.DELIVERED
-    elif status == "failure":
-        status = NotificationStatus.FAILED
-    # if status is not success or failure the client raised an exception and this method will retry
-
-    if status == NotificationStatus.DELIVERED:
-        sanitize_successful_notification_by_id(
-            notification_id, carrier=carrier, provider_response=provider_response
-        )
-        current_app.logger.info(
-            f"Sanitized notification {notification_id} that was successfully delivered"
-        )
-    else:
-        update_notification_status_by_id(
-            notification_id,
-            status,
-            carrier=carrier,
-            provider_response=provider_response,
-        )
-        current_app.logger.info(
-            f"Updated notification {notification_id} with response '{provider_response}'"
-        )
 
 
 @notify_celery.task(
@@ -127,17 +39,8 @@ def deliver_sms(self, notification_id):
                 ansi_green + f"AUTHENTICATION CODE: {notification.content}" + ansi_reset
             )
         # Code branches off to send_to_providers.py
-        message_id = send_to_providers.send_sms_to_provider(notification)
+        send_to_providers.send_sms_to_provider(notification)
 
-        # DEPRECATED
-        # We have to put it in UTC.  For other timezones, the delay
-        # will be ignored and it will fire immediately (although this probably only affects developer testing)
-        my_eta = utc_now() + timedelta(seconds=DELIVERY_RECEIPT_DELAY_IN_SECONDS)
-        check_sms_delivery_receipt.apply_async(
-            [message_id, notification_id, notification.created_at],
-            eta=my_eta,
-            queue=QueueNames.CHECK_SMS,
-        )
     except Exception as e:
         update_notification_status_by_id(
             notification_id,

app/celery/scheduled_tasks.py

@@ -1,16 +1,17 @@
 from datetime import timedelta
 
 from flask import current_app
-from sqlalchemy import between, select, union, update
+from sqlalchemy import between, select
 from sqlalchemy.exc import SQLAlchemyError
 
-from app import db, notify_celery, zendesk_client
+from app import notify_celery, zendesk_client
 from app.celery.tasks import (
     get_recipient_csv_and_template_and_sender_id,
     process_incomplete_jobs,
     process_job,
     process_row,
 )
+from app.clients.cloudwatch.aws_cloudwatch import AwsCloudwatchClient
 from app.config import QueueNames
 from app.dao.invited_org_user_dao import (
     delete_org_invitations_created_more_than_two_days_ago,
@@ -18,10 +19,14 @@ from app.dao.invited_org_user_dao import (
 from app.dao.invited_user_dao import expire_invitations_created_more_than_two_days_ago
 from app.dao.jobs_dao import (
     dao_set_scheduled_jobs_to_pending,
+    dao_update_job,
     find_jobs_with_missing_rows,
     find_missing_row_for_job,
 )
-from app.dao.notifications_dao import notifications_not_yet_sent
+from app.dao.notifications_dao import (
+    dao_update_delivery_receipts,
+    notifications_not_yet_sent,
+)
 from app.dao.services_dao import (
     dao_find_services_sending_to_tv_numbers,
     dao_find_services_with_high_failure_rates,
@@ -31,6 +36,7 @@ from app.enums import JobStatus, NotificationType
 from app.models import Job
 from app.notifications.process_notifications import send_notification_to_queue
 from app.utils import utc_now
+from notifications_utils import aware_utcnow
 from notifications_utils.clients.zendesk.zendesk_client import NotifySupportTicket
 
 MAX_NOTIFICATION_FAILS = 10000
@@ -94,40 +100,41 @@ def check_job_status():
     select
     from jobs
     where job_status == 'in progress'
-    and processing started between 30 and 35 minutes ago
+    and processing started some time ago
     OR where the job_status == 'pending'
-    and the job scheduled_for timestamp is between 30 and 35 minutes ago.
+    and the job scheduled_for timestamp is some time ago.
     if any results then
         update the job_status to 'error'
         process the rows in the csv that are missing (in another task) just do the check here.
     """
-    thirty_minutes_ago = utc_now() - timedelta(minutes=30)
+    START_MINUTES = 245
-    thirty_five_minutes_ago = utc_now() - timedelta(minutes=35)
+    END_MINUTES = 240
+    end_minutes_ago = utc_now() - timedelta(minutes=END_MINUTES)
+    start_minutes_ago = utc_now() - timedelta(minutes=START_MINUTES)
 
     incomplete_in_progress_jobs = select(Job).where(
         Job.job_status == JobStatus.IN_PROGRESS,
-        between(Job.processing_started, thirty_five_minutes_ago, thirty_minutes_ago),
+        between(Job.processing_started, start_minutes_ago, end_minutes_ago),
     )
     incomplete_pending_jobs = select(Job).where(
         Job.job_status == JobStatus.PENDING,
         Job.scheduled_for.isnot(None),
-        between(Job.scheduled_for, thirty_five_minutes_ago, thirty_minutes_ago),
+        between(Job.scheduled_for, start_minutes_ago, end_minutes_ago),
     )
 
-    jobs_not_complete_after_30_minutes = db.session.execute(
+    jobs_not_complete_after_allotted_time = (
-        union(incomplete_in_progress_jobs, incomplete_pending_jobs).order_by(
+        incomplete_in_progress_jobs.union(incomplete_pending_jobs)
-            Job.processing_started, Job.scheduled_for
+        .order_by(Job.processing_started, Job.scheduled_for)
-        )
+        .all()
-    ).all()
+    )
 
     # temporarily mark them as ERROR so that they don't get picked up by future check_job_status tasks
     # if they haven't been re-processed in time.
     job_ids = []
-    for job in jobs_not_complete_after_30_minutes:
+    for job in jobs_not_complete_after_allotted_time:
-        db.session.execute(
+        job.job_status = JobStatus.ERROR
-            update(Job).where(Job.id == job.id).values(job_status=JobStatus.ERROR)
+        dao_update_job(job)
-        )
+        job_ids.append(str(job.id))
-        db.session.commit()
 
         job_ids.append(str(job.id))
     if job_ids:
@@ -171,9 +178,7 @@ def check_for_missing_rows_in_completed_jobs():
         for row_to_process in missing_rows:
             row = recipient_csv[row_to_process.missing_row]
             current_app.logger.info(
-                "Processing missing row: {} for job: {}".format(
+                f"Processing missing row: {row_to_process.missing_row} for job: {job.id}"
-                    row_to_process.missing_row, job.id
-                )
             )
             process_row(row, template, job, job.service, sender_id=sender_id)
 
@@ -233,3 +238,45 @@ def check_for_services_with_high_failure_rates_or_sending_to_tv_numbers():
                 technical_ticket=True,
             )
             zendesk_client.send_ticket_to_zendesk(ticket)
+
+
+@notify_celery.task(
+    bind=True, max_retries=7, default_retry_delay=3600, name="process-delivery-receipts"
+)
+def process_delivery_receipts(self):
+    """
+    Every eight minutes or so (see config.py) we run this task, which searches the last ten
+    minutes of logs for delivery receipts and batch updates the db with the results.  The overlap
+    is intentional.  We don't mind re-updating things, it is better than losing data.
+
+    We also set this to retry with exponential backoff in the case of failure.  The only way this would
+    fail is if, for example the db went down, or redis filled causing the app to stop processing.  But if
+    it does fail, we need to go back over at some point when things are running again and process those results.
+    """
+    try:
+        batch_size = 1000  # in theory with postgresql this could be 10k to 20k?
+
+        cloudwatch = AwsCloudwatchClient()
+        cloudwatch.init_app(current_app)
+        start_time = aware_utcnow() - timedelta(minutes=10)
+        end_time = aware_utcnow()
+        delivered_receipts, failed_receipts = cloudwatch.check_delivery_receipts(
+            start_time, end_time
+        )
+        delivered_receipts = list(delivered_receipts)
+        for i in range(0, len(delivered_receipts), batch_size):
+            batch = delivered_receipts[i : i + batch_size]
+            dao_update_delivery_receipts(batch, True)
+        failed_receipts = list(failed_receipts)
+        for i in range(0, len(failed_receipts), batch_size):
+            batch = failed_receipts[i : i + batch_size]
+            dao_update_delivery_receipts(batch, False)
+    except Exception as ex:
+        retry_count = self.request.retries
+        wait_time = 3600 * 2**retry_count
+        try:
+            raise self.retry(ex=ex, countdown=wait_time)
+        except self.MaxRetriesExceededError:
+            current_app.logger.error(
+                "Failed process delivery receipts after max retries"
+            )

app/celery/tasks.py

@@ -1,4 +1,5 @@
 import json
+from time import sleep
 
 from celery.signals import task_postrun
 from flask import current_app
@@ -38,9 +39,7 @@ def process_job(job_id, sender_id=None):
     start = utc_now()
     job = dao_get_job_by_id(job_id)
     current_app.logger.info(
-        "Starting process-job task for job id {} with status: {}".format(
+        f"Starting process-job task for job id {job_id} with status: {job.job_status}"
-            job_id, job.job_status
-        )
     )
 
     if job.job_status != JobStatus.PENDING:
@@ -56,7 +55,7 @@ def process_job(job_id, sender_id=None):
         job.job_status = JobStatus.CANCELLED
         dao_update_job(job)
         current_app.logger.warning(
-            "Job {} has been cancelled, service {} is inactive".format(
+            f"Job {job_id} has been cancelled, service {service.id} is inactive".format(
                 job_id, service.id
             )
         )
@@ -70,13 +69,21 @@ def process_job(job_id, sender_id=None):
     )
 
     current_app.logger.info(
-        "Starting job {} processing {} notifications".format(
+        f"Starting job {job_id} processing {job.notification_count} notifications"
-            job_id, job.notification_count
-        )
     )
 
+    # notify-api-1495 we are going to sleep periodically to give other
+    # jobs running at the same time a chance to get some of their messages
+    # sent.  Sleep for 1 second after every 3 sends, which gives us throughput
+    # of about 3600*3 per hour and would keep the queue clear assuming only one sender.
+    # It will also hopefully eliminate throttling when we send messages which we are
+    # currently seeing.
+    count = 0
     for row in recipient_csv.get_rows():
         process_row(row, template, job, service, sender_id=sender_id)
+        count = count + 1
+        if count % 3 == 0:
+            sleep(1)
 
     # End point/Exit point for message send flow.
     job_complete(job, start=start)
@@ -206,9 +213,7 @@ def save_sms(self, service_id, notification_id, encrypted_notification, sender_i
                 f"service not allowed to send for job_id {notification.get('job', None)}, aborting"
             )
         )
-        current_app.logger.debug(
+        current_app.logger.debug(f"SMS {notification_id} failed as restricted service")
-            "SMS {} failed as restricted service".format(notification_id)
-        )
         return
 
     try:
@@ -218,22 +223,30 @@ def save_sms(self, service_id, notification_id, encrypted_notification, sender_i
             job = dao_get_job_by_id(job_id)
             created_by_id = job.created_by_id
 
-        saved_notification = persist_notification(
+        try:
-            template_id=notification["template"],
+            saved_notification = persist_notification(
-            template_version=notification["template_version"],
+                template_id=notification["template"],
-            recipient=notification["to"],
+                template_version=notification["template_version"],
-            service=service,
+                recipient=notification["to"],
-            personalisation=notification.get("personalisation"),
+                service=service,
-            notification_type=NotificationType.SMS,
+                personalisation=notification.get("personalisation"),
-            api_key_id=None,
+                notification_type=NotificationType.SMS,
-            key_type=KeyType.NORMAL,
+                api_key_id=None,
-            created_at=utc_now(),
+                key_type=KeyType.NORMAL,
-            created_by_id=created_by_id,
+                created_at=utc_now(),
-            job_id=notification.get("job", None),
+                created_by_id=created_by_id,
-            job_row_number=notification.get("row_number", None),
+                job_id=notification.get("job", None),
-            notification_id=notification_id,
+                job_row_number=notification.get("row_number", None),
-            reply_to_text=reply_to_text,
+                notification_id=notification_id,
-        )
+                reply_to_text=reply_to_text,
+            )
+        except IntegrityError:
+            current_app.logger.warning(
+                f"{NotificationType.SMS}: {notification_id} already exists."
+            )
+            # If we don't have the return statement here, we will fall through and end
+            # up retrying because IntegrityError is a subclass of SQLAlchemyError
+            return
 
         # Kick off sns process in provider_tasks.py
         sn = saved_notification
@@ -247,11 +260,8 @@ def save_sms(self, service_id, notification_id, encrypted_notification, sender_i
         )
 
         current_app.logger.debug(
-            "SMS {} created at {} for job {}".format(
+            f"SMS {saved_notification.id} created at {saved_notification.created_at} "
-                saved_notification.id,
+            f"for job {notification.get('job', None)}"
-                saved_notification.created_at,
-                notification.get("job", None),
-            )
         )
 
     except SQLAlchemyError as e:

app/clients/cloudwatch/aws_cloudwatch.py

@@ -1,15 +1,12 @@
 import json
 import os
 import re
-from datetime import timedelta
 
 from boto3 import client
 from flask import current_app
 
 from app.clients import AWS_CLIENT_CONFIG, Client
 from app.cloudfoundry_config import cloud_config
-from app.exceptions import NotificationTechnicalFailureException
-from app.utils import hilite, utc_now
 
 
 class AwsCloudwatchClient(Client):
@@ -49,48 +46,32 @@ class AwsCloudwatchClient(Client):
     def is_localstack(self):
         return self._is_localstack
 
-    def _get_log(self, my_filter, log_group_name, sent_at):
+    def _get_log(self, log_group_name, start, end):
         # Check all cloudwatch logs from the time the notification was sent (currently 5 minutes previously) until now
-        now = utc_now()
-        beginning = sent_at
         next_token = None
         all_log_events = []
-        current_app.logger.info(f"START TIME {beginning} END TIME {now}")
+
-        # There has been a change somewhere and the time range we were previously using has become too
-        # narrow or wrong in some way, so events can't be found.  For the time being, adjust by adding
-        # a buffer on each side of 12 hours.
-        TWELVE_HOURS = 12 * 60 * 60 * 1000
         while True:
             if next_token:
                 response = self._client.filter_log_events(
                     logGroupName=log_group_name,
-                    filterPattern=my_filter,
                     nextToken=next_token,
-                    startTime=int(beginning.timestamp() * 1000) - TWELVE_HOURS,
+                    startTime=int(start.timestamp() * 1000),
-                    endTime=int(now.timestamp() * 1000) + TWELVE_HOURS,
+                    endTime=int(end.timestamp() * 1000),
                 )
             else:
                 response = self._client.filter_log_events(
                     logGroupName=log_group_name,
-                    filterPattern=my_filter,
+                    startTime=int(start.timestamp() * 1000),
-                    startTime=int(beginning.timestamp() * 1000) - TWELVE_HOURS,
+                    endTime=int(end.timestamp() * 1000),
-                    endTime=int(now.timestamp() * 1000) + TWELVE_HOURS,
                 )
             log_events = response.get("events", [])
             all_log_events.extend(log_events)
-            if len(log_events) > 0:
-                # We found it
-
-                break
             next_token = response.get("nextToken")
             if not next_token:
                 break
         return all_log_events
 
-    def _extract_account_number(self, ses_domain_arn):
-        account_number = ses_domain_arn.split(":")
-        return account_number
-
     def warn_if_dev_is_opted_out(self, provider_response, notification_id):
         if (
             "is opted out" in provider_response.lower()
@@ -108,60 +89,108 @@ class AwsCloudwatchClient(Client):
                 return logline
         return None
 
-    def check_sms(self, message_id, notification_id, created_at):
+    def _extract_account_number(self, ses_domain_arn):
+        account_number = ses_domain_arn.split(":")
+        return account_number
+
+    def event_to_db_format(self, event):
+
+        # massage the data into the form the db expects.  When we switch
+        # from filter_log_events to log insights this will be convenient
+        if isinstance(event, str):
+            event = json.loads(event)
+
+        # Don't trust AWS to always send the same JSON structure back
+        # However, if we don't get message_id and status we might as well blow up
+        # because it's pointless to continue
+        phone_carrier = self._aws_value_or_default(event, "delivery", "phoneCarrier")
+        provider_response = self._aws_value_or_default(
+            event, "delivery", "providerResponse"
+        )
+        my_timestamp = self._aws_value_or_default(event, "notification", "timestamp")
+        return {
+            "notification.messageId": event["notification"]["messageId"],
+            "status": event["status"],
+            "delivery.phoneCarrier": phone_carrier,
+            "delivery.providerResponse": provider_response,
+            "@timestamp": my_timestamp,
+        }
+
+    # Here is an example of how to get the events with log insights
+    # def do_log_insights():
+    #     query = """
+    #     fields @timestamp, status, message, recipient
+    #     | filter status = "DELIVERED"
+    #     | sort @timestamp asc
+    #     """
+    #     temp_client = boto3.client(
+    #             "logs",
+    #             region_name="us-gov-west-1",
+    #             aws_access_key_id=AWS_ACCESS_KEY_ID,
+    #             aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
+    #             config=AWS_CLIENT_CONFIG,
+    #         )
+    #     start = utc_now()
+    #     end = utc_now - timedelta(hours=1)
+    #     response = temp_client.start_query(
+    #         logGroupName = LOG_GROUP_NAME_DELIVERED,
+    #         startTime = int(start.timestamp()),
+    #         endTime= int(end.timestamp()),
+    #         queryString = query
+
+    #     )
+    #     query_id = response['queryId']
+    #     while True:
+    #         result = temp_client.get_query_results(queryId=query_id)
+    #         if result['status'] == 'Complete':
+    #             break
+    #         time.sleep(1)
+
+    #     delivery_receipts = []
+    #     for log in result['results']:
+    #         receipt = {field['field']: field['value'] for field in log}
+    #         delivery_receipts.append(receipt)
+    #         print(receipt)
+
+    #     print(len(delivery_receipts))
+
+    # In the long run we want to use Log Insights because it is more efficient
+    # that filter_log_events.  But we are blocked by a permissions issue in the broker.
+    # So for now, use filter_log_events and grab all log_events over a 10 minute interval,
+    # and run this on a schedule.
+    def check_delivery_receipts(self, start, end):
         region = cloud_config.sns_region
-        # TODO this clumsy approach to getting the account number will be fixed as part of notify-api #258
         account_number = self._extract_account_number(cloud_config.ses_domain_arn)
-
-        time_now = utc_now()
         log_group_name = f"sns/{region}/{account_number[4]}/DirectPublishToPhoneNumber"
-        filter_pattern = '{$.notification.messageId="XXXXX"}'
+        delivered_event_set = self._get_receipts(log_group_name, start, end)
-        filter_pattern = filter_pattern.replace("XXXXX", message_id)
+        current_app.logger.info(
-        all_log_events = self._get_log(filter_pattern, log_group_name, created_at)
+            (f"Delivered message count: {len(delivered_event_set)}")
-        if all_log_events and len(all_log_events) > 0:
+        )
-            event = all_log_events[0]
-            message = json.loads(event["message"])
-            self.warn_if_dev_is_opted_out(
-                message["delivery"]["providerResponse"], notification_id
-            )
-            # Here we map the answer from aws to the message_id.
-            # Previously, in send_to_providers, we mapped the job_id and row number
-            # to the message id.  And on the admin side we mapped the csv filename
-            # to the job_id.  So by tracing through all the logs we can go:
-            # filename->job_id->message_id->what really happened
-            current_app.logger.info(
-                hilite(f"DELIVERED: {message} for message_id {message_id}")
-            )
-            return (
-                "success",
-                message["delivery"]["providerResponse"],
-                message["delivery"].get("phoneCarrier", "Unknown Carrier"),
-            )
-
         log_group_name = (
             f"sns/{region}/{account_number[4]}/DirectPublishToPhoneNumber/Failure"
         )
-        all_failed_events = self._get_log(filter_pattern, log_group_name, created_at)
+        failed_event_set = self._get_receipts(log_group_name, start, end)
-        if all_failed_events and len(all_failed_events) > 0:
+        current_app.logger.info((f"Failed message count: {len(failed_event_set)}"))
-            event = all_failed_events[0]
-            message = json.loads(event["message"])
-            self.warn_if_dev_is_opted_out(
-                message["delivery"]["providerResponse"], notification_id
-            )
 
-            current_app.logger.info(
+        return delivered_event_set, failed_event_set
-                hilite(f"FAILED: {message} for message_id {message_id}")
-            )
-            return (
-                "failure",
-                message["delivery"]["providerResponse"],
-                message["delivery"].get("phoneCarrier", "Unknown Carrier"),
-            )
 
-        if time_now > (created_at + timedelta(hours=3)):
+    def _get_receipts(self, log_group_name, start, end):
-            # see app/models.py Notification. This message corresponds to "permanent-failure",
+        event_set = set()
-            # but we are copy/pasting here to avoid circular imports.
+        all_events = self._get_log(log_group_name, start, end)
-            return "failure", "Unable to find carrier response."
+        for event in all_events:
-        raise NotificationTechnicalFailureException(
+            try:
-            f"No event found for message_id {message_id} notification_id {notification_id}"
+                actual_event = self.event_to_db_format(event["message"])
-        )
+                event_set.add(json.dumps(actual_event))
+            except Exception:
+                current_app.logger.exception(
+                    f"Could not format delivery receipt {event} for db insert"
+                )
+        return event_set
+
+    def _aws_value_or_default(self, event, top_level, second_level):
+        if event.get(top_level) is None or event[top_level].get(second_level) is None:
+            my_var = ""
+        else:
+            my_var = event[top_level][second_level]
+
+        return my_var

app/config.py

@@ -198,6 +198,11 @@ class Config(object):
                 "schedule": timedelta(minutes=63),
                 "options": {"queue": QueueNames.PERIODIC},
             },
+            "process-delivery-receipts": {
+                "task": "process-delivery-receipts",
+                "schedule": timedelta(minutes=8),
+                "options": {"queue": QueueNames.PERIODIC},
+            },
             "expire-or-delete-invitations": {
                 "task": "expire-or-delete-invitations",
                 "schedule": timedelta(minutes=66),

app/dao/notifications_dao.py

@@ -1,7 +1,20 @@
+import json
 from datetime import timedelta
 
 from flask import current_app
-from sqlalchemy import asc, delete, desc, func, or_, select, text, union, update
+from sqlalchemy import (
+    TIMESTAMP,
+    asc,
+    cast,
+    delete,
+    desc,
+    func,
+    or_,
+    select,
+    text,
+    union,
+    update,
+)
 from sqlalchemy.orm import joinedload
 from sqlalchemy.orm.exc import NoResultFound
 from sqlalchemy.sql import functions
@@ -53,6 +66,12 @@ def dao_get_last_date_template_was_used(template_id, service_id):
     return last_date
 
 
+def dao_notification_exists(notification_id) -> bool:
+    stmt = select(Notification).where(Notification.id == notification_id)
+    result = db.session.execute(stmt).scalar()
+    return result is not None
+
+
 @autocommit
 def dao_create_notification(notification):
     if not notification.id:
@@ -74,9 +93,7 @@ def dao_create_notification(notification):
     notification.normalised_to = "1"
 
     # notify-api-1454 insert only if it doesn't exist
-    stmt = select(Notification).where(Notification.id == notification.id)
+    if not dao_notification_exists(notification.id):
-    result = db.session.execute(stmt).scalar()
-    if result is None:
         db.session.add(notification)
 
 
@@ -725,3 +742,52 @@ def get_service_ids_with_notifications_on_date(notification_type, date):
             union(notification_table_query, ft_status_table_query).subquery()
         ).distinct()
     }
+
+
+def dao_update_delivery_receipts(receipts, delivered):
+    new_receipts = []
+    for r in receipts:
+        if isinstance(r, str):
+            r = json.loads(r)
+        new_receipts.append(r)
+
+    receipts = new_receipts
+
+    id_to_carrier = {
+        r["notification.messageId"]: r["delivery.phoneCarrier"] for r in receipts
+    }
+    id_to_provider_response = {
+        r["notification.messageId"]: r["delivery.providerResponse"] for r in receipts
+    }
+    id_to_timestamp = {r["notification.messageId"]: r["@timestamp"] for r in receipts}
+
+    status_to_update_with = NotificationStatus.DELIVERED
+    if not delivered:
+        status_to_update_with = NotificationStatus.FAILED
+    stmt = (
+        update(Notification)
+        .where(Notification.message_id.in_(id_to_carrier.keys()))
+        .values(
+            carrier=case(
+                *[
+                    (Notification.message_id == key, value)
+                    for key, value in id_to_carrier.items()
+                ]
+            ),
+            status=status_to_update_with,
+            sent_at=case(
+                *[
+                    (Notification.message_id == key, cast(value, TIMESTAMP))
+                    for key, value in id_to_timestamp.items()
+                ]
+            ),
+            provider_response=case(
+                *[
+                    (Notification.message_id == key, value)
+                    for key, value in id_to_provider_response.items()
+                ]
+            ),
+        )
+    )
+    db.session.execute(stmt)
+    db.session.commit()

app/notifications/process_notifications.py

@@ -8,6 +8,7 @@ from app.config import QueueNames
 from app.dao.notifications_dao import (
     dao_create_notification,
     dao_delete_notifications_by_id,
+    dao_notification_exists,
     get_notification_by_id,
 )
 from app.enums import KeyType, NotificationStatus, NotificationType
@@ -153,6 +154,10 @@ def persist_notification(
     return notification
 
 
+def notification_exists(notification_id):
+    return dao_notification_exists(notification_id)
+
+
 def send_notification_to_queue_detached(
     key_type, notification_type, notification_id, queue=None
 ):

tests/app/celery/test_provider_tasks.py

@@ -7,11 +7,7 @@ from celery.exceptions import MaxRetriesExceededError
 
 import app
 from app.celery import provider_tasks
-from app.celery.provider_tasks import (
+from app.celery.provider_tasks import deliver_email, deliver_sms
-    check_sms_delivery_receipt,
-    deliver_email,
-    deliver_sms,
-)
 from app.clients.email import EmailClientNonRetryableException
 from app.clients.email.aws_ses import (
     AwsSesClientException,
@@ -27,110 +23,10 @@ def test_should_have_decorated_tasks_functions():
     assert deliver_email.__wrapped__.__name__ == "deliver_email"
 
 
-def test_should_check_delivery_receipts_success(sample_notification, mocker):
-    mocker.patch("app.delivery.send_to_providers.send_sms_to_provider")
-    mocker.patch(
-        "app.celery.provider_tasks.aws_cloudwatch_client.is_localstack",
-        return_value=False,
-    )
-    mocker.patch(
-        "app.celery.provider_tasks.aws_cloudwatch_client.check_sms",
-        return_value=("success", "okay", "AT&T"),
-    )
-    mock_sanitize = mocker.patch(
-        "app.celery.provider_tasks.sanitize_successful_notification_by_id"
-    )
-    check_sms_delivery_receipt(
-        "message_id", sample_notification.id, "2024-10-20 00:00:00+0:00"
-    )
-    # This call should be made if the message was successfully delivered
-    mock_sanitize.assert_called_once()
-
-
-def test_should_check_delivery_receipts_failure(sample_notification, mocker):
-    mocker.patch("app.delivery.send_to_providers.send_sms_to_provider")
-    mocker.patch(
-        "app.celery.provider_tasks.aws_cloudwatch_client.is_localstack",
-        return_value=False,
-    )
-    mock_update = mocker.patch(
-        "app.celery.provider_tasks.update_notification_status_by_id"
-    )
-    mocker.patch(
-        "app.celery.provider_tasks.aws_cloudwatch_client.check_sms",
-        return_value=("failure", "not okay", "AT&T"),
-    )
-    mock_sanitize = mocker.patch(
-        "app.celery.provider_tasks.sanitize_successful_notification_by_id"
-    )
-    check_sms_delivery_receipt(
-        "message_id", sample_notification.id, "2024-10-20 00:00:00+0:00"
-    )
-    mock_sanitize.assert_not_called()
-    mock_update.assert_called_once()
-
-
-def test_should_check_delivery_receipts_client_error(sample_notification, mocker):
-    mocker.patch("app.delivery.send_to_providers.send_sms_to_provider")
-    mocker.patch(
-        "app.celery.provider_tasks.aws_cloudwatch_client.is_localstack",
-        return_value=False,
-    )
-    mock_update = mocker.patch(
-        "app.celery.provider_tasks.update_notification_status_by_id"
-    )
-    error_response = {"Error": {"Code": "SomeCode", "Message": "Some Message"}}
-    operation_name = "SomeOperation"
-    mocker.patch(
-        "app.celery.provider_tasks.aws_cloudwatch_client.check_sms",
-        side_effect=ClientError(error_response, operation_name),
-    )
-    mock_sanitize = mocker.patch(
-        "app.celery.provider_tasks.sanitize_successful_notification_by_id"
-    )
-    try:
-        check_sms_delivery_receipt(
-            "message_id", sample_notification.id, "2024-10-20 00:00:00+0:00"
-        )
-
-        assert 1 == 0
-    except ClientError:
-        mock_sanitize.assert_not_called()
-        mock_update.assert_called_once()
-
-
-def test_should_check_delivery_receipts_ntfe(sample_notification, mocker):
-    mocker.patch("app.delivery.send_to_providers.send_sms_to_provider")
-    mocker.patch(
-        "app.celery.provider_tasks.aws_cloudwatch_client.is_localstack",
-        return_value=False,
-    )
-    mock_update = mocker.patch(
-        "app.celery.provider_tasks.update_notification_status_by_id"
-    )
-    mocker.patch(
-        "app.celery.provider_tasks.aws_cloudwatch_client.check_sms",
-        side_effect=NotificationTechnicalFailureException(),
-    )
-    mock_sanitize = mocker.patch(
-        "app.celery.provider_tasks.sanitize_successful_notification_by_id"
-    )
-    try:
-        check_sms_delivery_receipt(
-            "message_id", sample_notification.id, "2024-10-20 00:00:00+0:00"
-        )
-
-        assert 1 == 0
-    except NotificationTechnicalFailureException:
-        mock_sanitize.assert_not_called()
-        mock_update.assert_called_once()
-
-
 def test_should_call_send_sms_to_provider_from_deliver_sms_task(
     sample_notification, mocker
 ):
     mocker.patch("app.delivery.send_to_providers.send_sms_to_provider")
-    mocker.patch("app.celery.provider_tasks.check_sms_delivery_receipt")
 
     deliver_sms(sample_notification.id)
     app.delivery.send_to_providers.send_sms_to_provider.assert_called_with(

tests/app/celery/test_scheduled_tasks.py

@@ -23,6 +23,8 @@ from notifications_utils.clients.zendesk.zendesk_client import NotifySupportTick
 from tests.app import load_example_csv
 from tests.app.db import create_job, create_notification, create_template
 
+CHECK_JOB_STATUS_TOO_OLD_MINUTES = 241
+
 
 def test_should_call_delete_codes_on_delete_verify_codes_task(
     notify_db_session, mocker
@@ -108,8 +110,9 @@ def test_check_job_status_task_calls_process_incomplete_jobs(mocker, sample_temp
     job = create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(minutes=31),
+        created_at=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
-        processing_started=utc_now() - timedelta(minutes=31),
+        processing_started=utc_now()
+        - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
         job_status=JobStatus.IN_PROGRESS,
     )
     create_notification(template=sample_template, job=job)
@@ -125,9 +128,10 @@ def test_check_job_status_task_calls_process_incomplete_jobs_when_scheduled_job_
     job = create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(hours=2),
+        created_at=utc_now() - timedelta(hours=5),
-        scheduled_for=utc_now() - timedelta(minutes=31),
+        scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
-        processing_started=utc_now() - timedelta(minutes=31),
+        processing_started=utc_now()
+        - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
         job_status=JobStatus.IN_PROGRESS,
     )
     check_job_status()
@@ -142,8 +146,8 @@ def test_check_job_status_task_calls_process_incomplete_jobs_for_pending_schedul
     job = create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(hours=2),
+        created_at=utc_now() - timedelta(hours=5),
-        scheduled_for=utc_now() - timedelta(minutes=31),
+        scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
         job_status=JobStatus.PENDING,
     )
 
@@ -175,17 +179,19 @@ def test_check_job_status_task_calls_process_incomplete_jobs_for_multiple_jobs(
     job = create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(hours=2),
+        created_at=utc_now() - timedelta(hours=5),
-        scheduled_for=utc_now() - timedelta(minutes=31),
+        scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
-        processing_started=utc_now() - timedelta(minutes=31),
+        processing_started=utc_now()
+        - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
         job_status=JobStatus.IN_PROGRESS,
     )
     job_2 = create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(hours=2),
+        created_at=utc_now() - timedelta(hours=5),
-        scheduled_for=utc_now() - timedelta(minutes=31),
+        scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
-        processing_started=utc_now() - timedelta(minutes=31),
+        processing_started=utc_now()
+        - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
         job_status=JobStatus.IN_PROGRESS,
     )
     check_job_status()
@@ -200,23 +206,24 @@ def test_check_job_status_task_only_sends_old_tasks(mocker, sample_template):
     job = create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(hours=2),
+        created_at=utc_now() - timedelta(hours=5),
-        scheduled_for=utc_now() - timedelta(minutes=31),
+        scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
-        processing_started=utc_now() - timedelta(minutes=31),
+        processing_started=utc_now()
+        - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
         job_status=JobStatus.IN_PROGRESS,
     )
     create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(minutes=31),
+        created_at=utc_now() - timedelta(minutes=300),
-        processing_started=utc_now() - timedelta(minutes=29),
+        processing_started=utc_now() - timedelta(minutes=239),
         job_status=JobStatus.IN_PROGRESS,
     )
     create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(minutes=50),
+        created_at=utc_now() - timedelta(minutes=300),
-        scheduled_for=utc_now() - timedelta(minutes=29),
+        scheduled_for=utc_now() - timedelta(minutes=239),
         job_status=JobStatus.PENDING,
     )
     check_job_status()
@@ -230,16 +237,17 @@ def test_check_job_status_task_sets_jobs_to_error(mocker, sample_template):
     job = create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(hours=2),
+        created_at=utc_now() - timedelta(hours=5),
-        scheduled_for=utc_now() - timedelta(minutes=31),
+        scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
-        processing_started=utc_now() - timedelta(minutes=31),
+        processing_started=utc_now()
+        - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
         job_status=JobStatus.IN_PROGRESS,
     )
     job_2 = create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(minutes=31),
+        created_at=utc_now() - timedelta(minutes=300),
-        processing_started=utc_now() - timedelta(minutes=29),
+        processing_started=utc_now() - timedelta(minutes=239),
         job_status=JobStatus.IN_PROGRESS,
     )
     check_job_status()
@@ -311,16 +319,18 @@ def test_check_job_status_task_does_not_raise_error(sample_template):
     create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(hours=2),
+        created_at=utc_now() - timedelta(hours=5),
-        scheduled_for=utc_now() - timedelta(minutes=31),
+        scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
-        processing_started=utc_now() - timedelta(minutes=31),
+        processing_started=utc_now()
+        - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
         job_status=JobStatus.FINISHED,
     )
     create_job(
         template=sample_template,
         notification_count=3,
-        created_at=utc_now() - timedelta(minutes=31),
+        created_at=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
-        processing_started=utc_now() - timedelta(minutes=31),
+        processing_started=utc_now()
+        - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
         job_status=JobStatus.FINISHED,
     )
 

tests/app/clients/test_aws_cloudwatch.py

@@ -1,8 +1,9 @@
+import json
+
 import pytest
 from flask import current_app
 
 from app import aws_cloudwatch_client
-from app.utils import utc_now
 
 
 def test_check_sms_no_event_error_condition(notify_api, mocker):
@@ -74,51 +75,6 @@ def test_warn_if_dev_is_opted_out(response, notify_id, expected_message):
     assert result == expected_message
 
 
-def test_check_sms_success(notify_api, mocker):
-    aws_cloudwatch_client.init_app(current_app)
-    boto_mock = mocker.patch.object(aws_cloudwatch_client, "_client", create=True)
-    boto_mock.filter_log_events.side_effect = side_effect
-    mocker.patch.dict(
-        "os.environ",
-        {"SES_DOMAIN_ARN": "arn:aws:ses:us-west-2:12345:identity/ses-xxx.xxx.xxx.xxx"},
-    )
-
-    message_id = "succeed"
-    notification_id = "ccc"
-    created_at = utc_now()
-    with notify_api.app_context():
-        aws_cloudwatch_client.check_sms(message_id, notification_id, created_at)
-
-    # We check the 'success' log group first and if we find the message_id, we are done, so there is only 1 call
-    assert boto_mock.filter_log_events.call_count == 1
-    mock_call = str(boto_mock.filter_log_events.mock_calls[0])
-    assert "Failure" not in mock_call
-    assert "succeed" in mock_call
-    assert "notification.messageId" in mock_call
-
-
-def test_check_sms_failure(notify_api, mocker):
-    aws_cloudwatch_client.init_app(current_app)
-    boto_mock = mocker.patch.object(aws_cloudwatch_client, "_client", create=True)
-    boto_mock.filter_log_events.side_effect = side_effect
-    mocker.patch.dict(
-        "os.environ",
-        {"SES_DOMAIN_ARN": "arn:aws:ses:us-west-2:12345:identity/ses-xxx.xxx.xxx.xxx"},
-    )
-    message_id = "fail"
-    notification_id = "bbb"
-    created_at = utc_now()
-    with notify_api.app_context():
-        aws_cloudwatch_client.check_sms(message_id, notification_id, created_at)
-
-    # We check the 'success' log group and find nothing, so we then check the 'fail' log group -- two calls.
-    assert boto_mock.filter_log_events.call_count == 2
-    mock_call = str(boto_mock.filter_log_events.mock_calls[1])
-    assert "Failure" in mock_call
-    assert "fail" in mock_call
-    assert "notification.messageId" in mock_call
-
-
 def test_extract_account_number_gov_cloud():
     domain_arn = "arn:aws-us-gov:ses:us-gov-west-1:12345:identity/ses-abc.xxx.xxx.xxx"
     actual_account_number = aws_cloudwatch_client._extract_account_number(domain_arn)
@@ -133,3 +89,65 @@ def test_extract_account_number_gov_staging():
     assert len(actual_account_number) == 6
     expected_account_number = "12345"
     assert actual_account_number[4] == expected_account_number
+
+
+def test_check_delivery_receipts():
+    pass
+
+
+def test_aws_value_or_default():
+    event = {
+        "delivery": {"phoneCarrier": "AT&T"},
+        "notification": {"timestamp": "2024-01-01T:12:00:00Z"},
+    }
+    assert (
+        aws_cloudwatch_client._aws_value_or_default(event, "delivery", "phoneCarrier")
+        == "AT&T"
+    )
+    assert (
+        aws_cloudwatch_client._aws_value_or_default(
+            event, "delivery", "providerResponse"
+        )
+        == ""
+    )
+    assert (
+        aws_cloudwatch_client._aws_value_or_default(event, "notification", "timestamp")
+        == "2024-01-01T:12:00:00Z"
+    )
+    assert (
+        aws_cloudwatch_client._aws_value_or_default(event, "nonexistent", "field") == ""
+    )
+
+
+def test_event_to_db_format_with_missing_fields():
+    event = {
+        "notification": {"messageId": "12345"},
+        "status": "UNKNOWN",
+        "delivery": {},
+    }
+    result = aws_cloudwatch_client.event_to_db_format(event)
+    assert result == {
+        "notification.messageId": "12345",
+        "status": "UNKNOWN",
+        "delivery.phoneCarrier": "",
+        "delivery.providerResponse": "",
+        "@timestamp": "",
+    }
+
+
+def test_event_to_db_format_with_string_input():
+    event = json.dumps(
+        {
+            "notification": {"messageId": "67890", "timestamp": "2024-01-01T14:00:00Z"},
+            "status": "FAILED",
+            "delivery": {"phoneCarrier": "Verizon", "providerResponse": "Error"},
+        }
+    )
+    result = aws_cloudwatch_client.event_to_db_format(event)
+    assert result == {
+        "notification.messageId": "67890",
+        "status": "FAILED",
+        "delivery.phoneCarrier": "Verizon",
+        "delivery.providerResponse": "Error",
+        "@timestamp": "2024-01-01T14:00:00Z",
+    }

tests/app/dao/notification_dao/test_notification_dao.py

@@ -1,6 +1,7 @@
 import uuid
 from datetime import date, datetime, timedelta
 from functools import partial
+from unittest.mock import MagicMock, patch
 
 import pytest
 from freezegun import freeze_time
@@ -19,6 +20,7 @@ from app.dao.notifications_dao import (
     dao_get_notification_history_by_reference,
     dao_get_notifications_by_recipient_or_reference,
     dao_timeout_notifications,
+    dao_update_delivery_receipts,
     dao_update_notification,
     dao_update_notifications_by_reference,
     get_notification_by_id,
@@ -2000,6 +2002,34 @@ def test_notifications_not_yet_sent_return_no_rows(sample_service, notification_
     assert len(results) == 0
 
 
+def test_update_delivery_receipts(mocker):
+    mock_session = mocker.patch("app.dao.notifications_dao.db.session")
+    receipts = [
+        '{"notification.messageId": "msg1", "delivery.phoneCarrier": "carrier1", "delivery.providerResponse": "resp1", "@timestamp": "2024-01-01T12:00:00"}',  # noqa
+        '{"notification.messageId": "msg2", "delivery.phoneCarrier": "carrier2", "delivery.providerResponse": "resp2", "@timestamp": "2024-01-01T13:00:00"}',  # noqa
+    ]
+    delivered = True
+    mock_update = MagicMock()
+    mock_where = MagicMock()
+    mock_values = MagicMock()
+    mock_update.where.return_value = mock_where
+    mock_where.values.return_value = mock_values
+
+    mock_session.execute.return_value = None
+    with patch("app.dao.notifications_dao.update", return_value=mock_update):
+        dao_update_delivery_receipts(receipts, delivered)
+    mock_update.where.assert_called_once()
+    mock_where.values.assert_called_once()
+    mock_session.execute.assert_called_once_with(mock_values)
+    mock_session.commit.assert_called_once()
+
+    args, kwargs = mock_where.values.call_args
+    assert "carrier" in kwargs
+    assert "status" in kwargs
+    assert "sent_at" in kwargs
+    assert "provider_response" in kwargs
+
+
 @pytest.mark.parametrize(
     "created_at_utc,date_to_check,expected_count",
     [