Merge pull request #1522 from GSA/notify-api-1520

cleanup pending notifications

app/celery/scheduled_tasks.py

@@ -24,6 +24,7 @@ from app.dao.jobs_dao import (
     find_missing_row_for_job,
 )
 from app.dao.notifications_dao import (
+    dao_close_out_delivery_receipts,
     dao_update_delivery_receipts,
     notifications_not_yet_sent,
 )
@@ -278,3 +279,10 @@ def process_delivery_receipts(self):
             current_app.logger.error(
                 "Failed process delivery receipts after max retries"
             )
+
+
+@notify_celery.task(
+    bind=True, max_retries=2, default_retry_delay=3600, name="cleanup-delivery-receipts"
+)
+def cleanup_delivery_receipts(self):
+    dao_close_out_delivery_receipts()

app/config.py

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

app/dao/jobs_dao.py

@@ -45,7 +45,7 @@ def dao_get_job_by_service_id_and_job_id(service_id, job_id):
 
 def dao_get_unfinished_jobs():
     stmt = select(Job).filter(Job.processing_finished.is_(None))
-    return db.session.execute(stmt).all()
+    return db.session.execute(stmt).scalars().all()
 
 
 def dao_get_jobs_by_service_id(

app/dao/notifications_dao.py

@@ -780,3 +780,22 @@ def dao_update_delivery_receipts(receipts, delivered):
         f"#loadtestperformance batch update query time: \
         updated {len(receipts)} notification in {elapsed_time} ms"
     )
+
+
+def dao_close_out_delivery_receipts():
+    THREE_DAYS_AGO = utc_now() - timedelta(minutes=3)
+    stmt = (
+        update(Notification)
+        .where(
+            Notification.status == NotificationStatus.PENDING,
+            Notification.sent_at < THREE_DAYS_AGO,
+        )
+        .values(status=NotificationStatus.FAILED, provider_response="Technical Failure")
+    )
+    result = db.session.execute(stmt)
+
+    db.session.commit()
+    if result:
+        current_app.logger.info(
+            f"Marked {result.rowcount} notifications as technical failures"
+        )

tests/app/dao/notification_dao/test_notification_dao.py

@@ -11,6 +11,7 @@ from sqlalchemy.orm.exc import NoResultFound
 
 from app import db
 from app.dao.notifications_dao import (
+    dao_close_out_delivery_receipts,
     dao_create_notification,
     dao_delete_notifications_by_id,
     dao_get_last_notification_added_for_job_id,
@@ -2026,6 +2027,23 @@ def test_update_delivery_receipts(mocker):
     assert "provider_response" in kwargs
 
 
+def test_close_out_delivery_receipts(mocker):
+    mock_session = mocker.patch("app.dao.notifications_dao.db.session")
+    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_close_out_delivery_receipts()
+    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()
+
+
 @pytest.mark.parametrize(
     "created_at_utc,date_to_check,expected_count",
     [