From 874c8ffb541a7112c0347e02caa6b876e5d9a75a Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 07:58:24 -0800
Subject: [PATCH 01/40] try batch inserts

---
 app/celery/scheduled_tasks.py              | 18 +++++++++++++++++-
 app/config.py                              |  5 +++++
 app/dao/notifications_dao.py               | 10 ++++++++++
 app/notifications/process_notifications.py |  8 ++++----
 4 files changed, 36 insertions(+), 5 deletions(-)

diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py
index cb0e0886e..e173c923a 100644
--- a/app/celery/scheduled_tasks.py
+++ b/app/celery/scheduled_tasks.py
@@ -1,10 +1,11 @@
+import json
 from datetime import timedelta
 
 from flask import current_app
 from sqlalchemy import between
 from sqlalchemy.exc import SQLAlchemyError
 
-from app import notify_celery, zendesk_client
+from app import notify_celery, redis_store, zendesk_client
 from app.celery.tasks import (
     get_recipient_csv_and_template_and_sender_id,
     process_incomplete_jobs,
@@ -24,6 +25,7 @@ from app.dao.jobs_dao import (
     find_missing_row_for_job,
 )
 from app.dao.notifications_dao import (
+    dao_batch_insert_notifications,
     dao_close_out_delivery_receipts,
     dao_update_delivery_receipts,
     notifications_not_yet_sent,
@@ -286,3 +288,17 @@ def process_delivery_receipts(self):
 )
 def cleanup_delivery_receipts(self):
     dao_close_out_delivery_receipts()
+
+
+@notify_celery.task(bind=True, name="batch-insert-notifications")
+def batch_insert_notifications(self):
+    batch = []
+    with redis_store.pipeline:
+        notification = redis_store.lpop("notification_queue")
+        batch.append(json.loads(notification))
+    try:
+        dao_batch_insert_notifications(batch)
+    except Exception as e:
+        for msg in batch:
+            redis_store.rpush("notification_queue", json.dumps(msg))
+        current_app.logger.exception(f"Notification batch insert failed {e}")
diff --git a/app/config.py b/app/config.py
index 580495731..bd19ffa59 100644
--- a/app/config.py
+++ b/app/config.py
@@ -208,6 +208,11 @@ class Config(object):
                 "schedule": timedelta(minutes=82),
                 "options": {"queue": QueueNames.PERIODIC},
             },
+            "batch-insert-notifications": {
+                "task": "batch-insert-notifications",
+                "schedule": 10.0,
+                "options": {"queue": QueueNames.PERIODIC},
+            },
             "expire-or-delete-invitations": {
                 "task": "expire-or-delete-invitations",
                 "schedule": timedelta(minutes=66),
diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py
index c8f2797a0..cd3c0e1aa 100644
--- a/app/dao/notifications_dao.py
+++ b/app/dao/notifications_dao.py
@@ -2,6 +2,7 @@ import json
 from datetime import timedelta
 from time import time
 
+import sqlalchemy
 from flask import current_app
 from sqlalchemy import (
     TIMESTAMP,
@@ -799,3 +800,12 @@ def dao_close_out_delivery_receipts():
         current_app.logger.info(
             f"Marked {result.rowcount} notifications as technical failures"
         )
+
+
+def dao_batch_insert_notifications(batch):
+    try:
+        db.session.bulk_save_objects(Notification(**msg) for msg in batch)
+        db.session.commit()
+        return len(batch)
+    except sqlalchemy.exc.SQLAlchemyError as e:
+        current_app.logger.exception(f"Error during batch insert {e}")
diff --git a/app/notifications/process_notifications.py b/app/notifications/process_notifications.py
index 5f1c6676d..347d2fc0b 100644
--- a/app/notifications/process_notifications.py
+++ b/app/notifications/process_notifications.py
@@ -6,7 +6,6 @@ from app import redis_store
 from app.celery import provider_tasks
 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,
@@ -139,8 +138,9 @@ def persist_notification(
 
     # if simulated create a Notification model to return but do not persist the Notification to the dB
     if not simulated:
-        current_app.logger.info("Firing dao_create_notification")
-        dao_create_notification(notification)
+        # current_app.logger.info("Firing dao_create_notification")
+        # dao_create_notification(notification)
+        redis_store.rpush("message_queue", notification)
         if key_type != KeyType.TEST and current_app.config["REDIS_ENABLED"]:
             current_app.logger.info(
                 "Redis enabled, querying cache key for service id: {}".format(
@@ -172,7 +172,7 @@ def send_notification_to_queue_detached(
         deliver_task = provider_tasks.deliver_email
 
     try:
-        deliver_task.apply_async([str(notification_id)], queue=queue)
+        deliver_task.apply_async([str(notification_id)], queue=queue, countdown=30)
     except Exception:
         dao_delete_notifications_by_id(notification_id)
         raise

From bbf5bace208bfb85986a160ac26926a1dfc489a7 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 08:05:51 -0800
Subject: [PATCH 02/40] add lpop and rpush to notify redis

---
 notifications_utils/clients/redis/redis_client.py | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/notifications_utils/clients/redis/redis_client.py b/notifications_utils/clients/redis/redis_client.py
index 1723dd2c1..3404d27e7 100644
--- a/notifications_utils/clients/redis/redis_client.py
+++ b/notifications_utils/clients/redis/redis_client.py
@@ -156,6 +156,14 @@ class RedisClient:
 
         return None
 
+    def rpush(self, key, value):
+        if self.active:
+            self.redis_store.rpush(key, value)
+
+    def lpop(self, key, value):
+        if self.active:
+            self.redis_store.lpop(key, value)
+
     def delete(self, *keys, raise_exception=False):
         keys = [prepare_value(k) for k in keys]
         if self.active:

From 64a61f5d362560427a0269e4f9e5c54eff02ffce Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 11:21:39 -0800
Subject: [PATCH 03/40] cleanup redis commands and flow

---
 app/celery/scheduled_tasks.py                 | 36 ++++++++++++++++---
 app/dao/notifications_dao.py                  |  4 ++-
 app/models.py                                 | 25 ++++++++++++-
 app/notifications/process_notifications.py    | 18 +++++-----
 .../clients/redis/redis_client.py             | 12 +++++--
 5 files changed, 77 insertions(+), 18 deletions(-)

diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py
index e173c923a..9fcfeeb04 100644
--- a/app/celery/scheduled_tasks.py
+++ b/app/celery/scheduled_tasks.py
@@ -36,7 +36,7 @@ from app.dao.services_dao import (
 )
 from app.dao.users_dao import delete_codes_older_created_more_than_a_day_ago
 from app.enums import JobStatus, NotificationType
-from app.models import Job
+from app.models import Job, Notification
 from app.notifications.process_notifications import send_notification_to_queue
 from app.utils import utc_now
 from notifications_utils import aware_utcnow
@@ -292,13 +292,39 @@ def cleanup_delivery_receipts(self):
 
 @notify_celery.task(bind=True, name="batch-insert-notifications")
 def batch_insert_notifications(self):
+    current_app.logger.info("ENTER SCHEDULED TASK")
     batch = []
-    with redis_store.pipeline:
-        notification = redis_store.lpop("notification_queue")
-        batch.append(json.loads(notification))
+    # with redis_store.pipeline():
+    #     while redis_store.llen("message_queue") > 0:
+    #         redis_store.lpop("message_queue")
+    #     current_app.logger.info("EMPTY!")
+    #     return
+    with redis_store.pipeline():
+        current_app.logger.info("PIPELINE")
+        # since this list is always growing, just grab what is available when
+        # this call is made and process that.
+        current_len = redis_store.llen("message_queue")
+        count = 0
+        while count < current_len:
+            count = count + 1
+            notification_bytes = redis_store.lpop("message_queue")
+            notification_dict = json.loads(notification_bytes.decode("utf-8"))
+            notification_dict["status"] = notification_dict.pop("notification_status")
+            notification_dict["created_at"] = utc_now()
+            notification = Notification(**notification_dict)
+            current_app.logger.info(
+                f"WHAT IS THIS NOTIFICATION {type(notification)} {notification}"
+            )
+            if notification is not None:
+                current_app.logger.info(
+                    f"SCHEDULED adding notification {notification.id} to batch"
+                )
+                batch.append(notification)
     try:
+        current_app.logger.info("GOING TO DO BATCH INSERT")
         dao_batch_insert_notifications(batch)
     except Exception as e:
+        current_app.logger.exception(f"Notification batch insert failed {e}")
+
         for msg in batch:
             redis_store.rpush("notification_queue", json.dumps(msg))
-        current_app.logger.exception(f"Notification batch insert failed {e}")
diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py
index cd3c0e1aa..92dcc234c 100644
--- a/app/dao/notifications_dao.py
+++ b/app/dao/notifications_dao.py
@@ -803,9 +803,11 @@ def dao_close_out_delivery_receipts():
 
 
 def dao_batch_insert_notifications(batch):
+    current_app.logger.info("DOING BATCH INSERT IN DAO")
     try:
-        db.session.bulk_save_objects(Notification(**msg) for msg in batch)
+        db.session.bulk_save_objects(batch)
         db.session.commit()
+        current_app.logger.info(f"SUCCESSFULLY INSERTED: {len(batch)}")
         return len(batch)
     except sqlalchemy.exc.SQLAlchemyError as e:
         current_app.logger.exception(f"Error during batch insert {e}")
diff --git a/app/models.py b/app/models.py
index fc7b855e4..ff734f8bf 100644
--- a/app/models.py
+++ b/app/models.py
@@ -5,7 +5,7 @@ from flask import current_app, url_for
 from sqlalchemy import CheckConstraint, Index, UniqueConstraint
 from sqlalchemy.dialects.postgresql import JSON, JSONB, UUID
 from sqlalchemy.ext.associationproxy import association_proxy
-from sqlalchemy.ext.declarative import declared_attr
+from sqlalchemy.ext.declarative import DeclarativeMeta, declared_attr
 from sqlalchemy.orm import validates
 from sqlalchemy.orm.collections import attribute_mapped_collection
 
@@ -1694,6 +1694,29 @@ class Notification(db.Model):
         else:
             return None
 
+    def serialize_for_redis(self, obj):
+        if isinstance(obj.__class__, DeclarativeMeta):
+            fields = {}
+            for column in obj.__table__.columns:
+                if column.name == "notification_status":
+                    new_name = "status"
+                    value = getattr(obj, new_name)
+                elif column.name == "created_at":
+                    value = (obj.created_at.strftime("%Y-%m-%d %H:%M:%S"),)
+                elif column.name in ["sent_at", "completed_at"]:
+                    value = None
+                elif column.name.endswith("_id"):
+                    value = getattr(obj, column.name)
+                    value = str(value)
+                else:
+                    value = getattr(obj, column.name)
+                if column.name in ["message_id", "api_key_id"]:
+                    pass  # do nothing because we don't have the message id yet
+                else:
+                    fields[column.name] = value
+            return fields
+        raise ValueError("Provided object is not a SQLAlchemy instance")
+
     def serialize_for_csv(self):
         serialized = {
             "row_number": (
diff --git a/app/notifications/process_notifications.py b/app/notifications/process_notifications.py
index 347d2fc0b..2be547f7a 100644
--- a/app/notifications/process_notifications.py
+++ b/app/notifications/process_notifications.py
@@ -1,3 +1,4 @@
+import json
 import uuid
 
 from flask import current_app
@@ -10,7 +11,7 @@ from app.dao.notifications_dao import (
     dao_notification_exists,
     get_notification_by_id,
 )
-from app.enums import KeyType, NotificationStatus, NotificationType
+from app.enums import NotificationStatus, NotificationType
 from app.errors import BadRequestError
 from app.models import Notification
 from app.utils import hilite, utc_now
@@ -140,16 +141,15 @@ def persist_notification(
     if not simulated:
         # current_app.logger.info("Firing dao_create_notification")
         # dao_create_notification(notification)
-        redis_store.rpush("message_queue", notification)
-        if key_type != KeyType.TEST and current_app.config["REDIS_ENABLED"]:
-            current_app.logger.info(
-                "Redis enabled, querying cache key for service id: {}".format(
-                    service.id
-                )
-            )
+        current_app.logger.info(
+            f"QUEUE LENTGH BEFOE {redis_store.llen("message_queue")}"
+        )
+        redis_store.rpush(
+            "message_queue", json.dumps(notification.serialize_for_redis(notification))
+        )
 
         current_app.logger.info(
-            f"{notification_type} {notification_id} created at {notification_created_at}"
+            f"QUEUE LENTGH AFTA {redis_store.llen("message_queue")}"
         )
     return notification
 
diff --git a/notifications_utils/clients/redis/redis_client.py b/notifications_utils/clients/redis/redis_client.py
index 3404d27e7..c41318243 100644
--- a/notifications_utils/clients/redis/redis_client.py
+++ b/notifications_utils/clients/redis/redis_client.py
@@ -38,6 +38,10 @@ class RedisClient:
     active = False
     scripts = {}
 
+    @classmethod
+    def pipeline(cls):
+        return cls.redis_store.pipeline()
+
     def init_app(self, app):
         self.active = app.config.get("REDIS_ENABLED")
         if self.active:
@@ -160,9 +164,13 @@ class RedisClient:
         if self.active:
             self.redis_store.rpush(key, value)
 
-    def lpop(self, key, value):
+    def lpop(self, key):
         if self.active:
-            self.redis_store.lpop(key, value)
+            return self.redis_store.lpop(key)
+
+    def llen(self, key):
+        if self.active:
+            return self.redis_store.llen(key)
 
     def delete(self, *keys, raise_exception=False):
         keys = [prepare_value(k) for k in keys]

From 5f7089fea04ae7016bbf86ebad9a60ae2ae14b3c Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 11:51:42 -0800
Subject: [PATCH 04/40] add countdown of 30 seconds for deliveries

---
 app/celery/tasks.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/app/celery/tasks.py b/app/celery/tasks.py
index 3743aa294..4086f684a 100644
--- a/app/celery/tasks.py
+++ b/app/celery/tasks.py
@@ -256,7 +256,7 @@ def save_sms(self, service_id, notification_id, encrypted_notification, sender_i
             )
         )
         provider_tasks.deliver_sms.apply_async(
-            [str(saved_notification.id)], queue=QueueNames.SEND_SMS
+            [str(saved_notification.id)], queue=QueueNames.SEND_SMS, countdown=30
         )
 
         current_app.logger.debug(

From 1fbe4277864f5b510042c3c1970ca95c21aa055d Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 12:03:00 -0800
Subject: [PATCH 05/40] revert behavior for emails, only sms needs optimization

---
 app/notifications/process_notifications.py | 19 ++++++++-----------
 1 file changed, 8 insertions(+), 11 deletions(-)

diff --git a/app/notifications/process_notifications.py b/app/notifications/process_notifications.py
index 2be547f7a..f6feca539 100644
--- a/app/notifications/process_notifications.py
+++ b/app/notifications/process_notifications.py
@@ -7,6 +7,7 @@ from app import redis_store
 from app.celery import provider_tasks
 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,
@@ -139,18 +140,14 @@ def persist_notification(
 
     # if simulated create a Notification model to return but do not persist the Notification to the dB
     if not simulated:
-        # current_app.logger.info("Firing dao_create_notification")
-        # dao_create_notification(notification)
-        current_app.logger.info(
-            f"QUEUE LENTGH BEFOE {redis_store.llen("message_queue")}"
-        )
-        redis_store.rpush(
-            "message_queue", json.dumps(notification.serialize_for_redis(notification))
-        )
+        if notification.notification_type == NotificationType.SMS:
+            redis_store.rpush(
+                "message_queue",
+                json.dumps(notification.serialize_for_redis(notification)),
+            )
+        else:
+            dao_create_notification(notification)
 
-        current_app.logger.info(
-            f"QUEUE LENTGH AFTA {redis_store.llen("message_queue")}"
-        )
     return notification
 
 

From 833146e4242a6ab2c08237b00cfa849a98f888bf Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 13:15:42 -0800
Subject: [PATCH 06/40] fix tests

---
 tests/app/celery/test_scheduled_tasks.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index f436aacf2..0c285ea94 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -308,10 +308,10 @@ def test_replay_created_notifications(notify_db_session, sample_service, mocker)
 
     replay_created_notifications()
     email_delivery_queue.assert_called_once_with(
-        [str(old_email.id)], queue="send-email-tasks"
+        [str(old_email.id)], queue="send-email-tasks", countdown=30
     )
     sms_delivery_queue.assert_called_once_with(
-        [str(old_sms.id)], queue="send-sms-tasks"
+        [str(old_sms.id)], queue="send-sms-tasks", countdown=30
     )
 
 

From 1eea4bb35b2ca2ea427fd5ffc99d33e44e553c69 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 13:38:36 -0800
Subject: [PATCH 07/40] fix tests

---
 tests/app/celery/test_tasks.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tests/app/celery/test_tasks.py b/tests/app/celery/test_tasks.py
index 4fccfb8cb..e77b64062 100644
--- a/tests/app/celery/test_tasks.py
+++ b/tests/app/celery/test_tasks.py
@@ -13,6 +13,7 @@ from sqlalchemy.exc import SQLAlchemyError
 
 from app import db, encryption
 from app.celery import provider_tasks, tasks
+from app.celery.scheduled_tasks import batch_insert_notifications
 from app.celery.tasks import (
     get_recipient_csv_and_template_and_sender_id,
     process_incomplete_job,
@@ -944,7 +945,7 @@ def test_save_sms_uses_sms_sender_reply_to_text(mocker, notify_db_session):
         notification_id,
         encryption.encrypt(notification),
     )
-
+    batch_insert_notifications()
     persisted_notification = Notification.query.one()
     assert persisted_notification.reply_to_text == "+12028675309"
 

From 9685b09677759e1fca73177b01b4c302403fa971 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 13:51:35 -0800
Subject: [PATCH 08/40] fix tests

---
 app/models.py                  | 3 +++
 tests/app/celery/test_tasks.py | 1 +
 2 files changed, 4 insertions(+)

diff --git a/app/models.py b/app/models.py
index ff734f8bf..50c47ec76 100644
--- a/app/models.py
+++ b/app/models.py
@@ -1714,6 +1714,9 @@ class Notification(db.Model):
                     pass  # do nothing because we don't have the message id yet
                 else:
                     fields[column.name] = value
+            current_app.logger.warning(f"FIELDS {fields}")
+            print(f"FIELDS {fields}", flush=True)
+
             return fields
         raise ValueError("Provided object is not a SQLAlchemy instance")
 
diff --git a/tests/app/celery/test_tasks.py b/tests/app/celery/test_tasks.py
index e77b64062..292879f9a 100644
--- a/tests/app/celery/test_tasks.py
+++ b/tests/app/celery/test_tasks.py
@@ -945,6 +945,7 @@ def test_save_sms_uses_sms_sender_reply_to_text(mocker, notify_db_session):
         notification_id,
         encryption.encrypt(notification),
     )
+
     batch_insert_notifications()
     persisted_notification = Notification.query.one()
     assert persisted_notification.reply_to_text == "+12028675309"

From 6bd044e684676c339c3b50b098f6b6426f72f72d Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 14:03:07 -0800
Subject: [PATCH 09/40] fix uuid

---
 tests/app/celery/test_tasks.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/app/celery/test_tasks.py b/tests/app/celery/test_tasks.py
index 292879f9a..40fe55cf6 100644
--- a/tests/app/celery/test_tasks.py
+++ b/tests/app/celery/test_tasks.py
@@ -939,7 +939,7 @@ def test_save_sms_uses_sms_sender_reply_to_text(mocker, notify_db_session):
     notification = _notification_json(template, to="2028675301")
     mocker.patch("app.celery.provider_tasks.deliver_sms.apply_async")
 
-    notification_id = uuid.uuid4()
+    notification_id = str(uuid.uuid4())
     save_sms(
         service.id,
         notification_id,

From 302e3ee79831a99b5b4bbb2bbc5cce2a31cc2eae Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 14:12:45 -0800
Subject: [PATCH 10/40] fix uuid

---
 app/notifications/process_notifications.py | 13 +++++++++----
 1 file changed, 9 insertions(+), 4 deletions(-)

diff --git a/app/notifications/process_notifications.py b/app/notifications/process_notifications.py
index f6feca539..02eb1f766 100644
--- a/app/notifications/process_notifications.py
+++ b/app/notifications/process_notifications.py
@@ -1,4 +1,5 @@
 import json
+import os
 import uuid
 
 from flask import current_app
@@ -141,10 +142,14 @@ def persist_notification(
     # if simulated create a Notification model to return but do not persist the Notification to the dB
     if not simulated:
         if notification.notification_type == NotificationType.SMS:
-            redis_store.rpush(
-                "message_queue",
-                json.dumps(notification.serialize_for_redis(notification)),
-            )
+            # it's just too hard with redis and timing to test this here
+            if os.getenv("NOTIFY_ENVIRONMENT") == "test":
+                dao_create_notification(notification)
+            else:
+                redis_store.rpush(
+                    "message_queue",
+                    json.dumps(notification.serialize_for_redis(notification)),
+                )
         else:
             dao_create_notification(notification)
 

From 6f7c7d2d667743623a93d549fec6dafb82d96558 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 14:23:03 -0800
Subject: [PATCH 11/40] fix uuid

---
 tests/app/celery/test_tasks.py                       | 10 +++++-----
 tests/app/notifications/test_process_notification.py |  3 ++-
 2 files changed, 7 insertions(+), 6 deletions(-)

diff --git a/tests/app/celery/test_tasks.py b/tests/app/celery/test_tasks.py
index 40fe55cf6..77641f10c 100644
--- a/tests/app/celery/test_tasks.py
+++ b/tests/app/celery/test_tasks.py
@@ -599,7 +599,7 @@ def test_should_save_sms_template_to_and_persist_with_job_id(sample_job, mocker)
     assert persisted_notification.notification_type == NotificationType.SMS
 
     provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-sms-tasks"
+        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=30
     )
 
 
@@ -670,7 +670,7 @@ def test_should_use_email_template_and_persist(
     assert persisted_notification.notification_type == NotificationType.EMAIL
 
     provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-email-tasks"
+        [str(persisted_notification.id)], queue="send-email-tasks", countdown=30
     )
 
 
@@ -707,7 +707,7 @@ def test_save_email_should_use_template_version_from_job_not_latest(
     assert not persisted_notification.sent_by
     assert persisted_notification.notification_type == NotificationType.EMAIL
     provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-email-tasks"
+        [str(persisted_notification.id)], queue="send-email-tasks", countdown=30
     )
 
 
@@ -738,7 +738,7 @@ def test_should_use_email_template_subject_placeholders(
     assert not persisted_notification.reference
     assert persisted_notification.notification_type == NotificationType.EMAIL
     provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-email-tasks"
+        [str(persisted_notification.id)], queue="send-email-tasks", countdown=30
     )
 
 
@@ -822,7 +822,7 @@ def test_should_use_email_template_and_persist_without_personalisation(
     assert not persisted_notification.reference
     assert persisted_notification.notification_type == NotificationType.EMAIL
     provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-email-tasks"
+        [str(persisted_notification.id)], queue="send-email-tasks", countdown=30
     )
 
 
diff --git a/tests/app/notifications/test_process_notification.py b/tests/app/notifications/test_process_notification.py
index 9f393b440..296f68adf 100644
--- a/tests/app/notifications/test_process_notification.py
+++ b/tests/app/notifications/test_process_notification.py
@@ -263,7 +263,7 @@ def test_send_notification_to_queue(
 
     send_notification_to_queue(notification=notification, queue=requested_queue)
 
-    mocked.assert_called_once_with([str(notification.id)], queue=expected_queue)
+    mocked.assert_called_once_with([str(notification.id)], queue=expected_queue, countdown=30)
 
 
 def test_send_notification_to_queue_throws_exception_deletes_notification(
@@ -278,6 +278,7 @@ def test_send_notification_to_queue_throws_exception_deletes_notification(
     mocked.assert_called_once_with(
         [(str(sample_notification.id))],
         queue="send-sms-tasks",
+        countdown=30
     )
 
     assert _get_notification_query_count() == 0

From c6d098743d6d15b3246bd48f6269f96f0c3bc5f2 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 14:35:03 -0800
Subject: [PATCH 12/40] fix uuid

---
 tests/app/celery/test_tasks.py             | 4 ++--
 tests/app/organization/test_invite_rest.py | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/tests/app/celery/test_tasks.py b/tests/app/celery/test_tasks.py
index 77641f10c..d4081fc97 100644
--- a/tests/app/celery/test_tasks.py
+++ b/tests/app/celery/test_tasks.py
@@ -738,7 +738,7 @@ def test_should_use_email_template_subject_placeholders(
     assert not persisted_notification.reference
     assert persisted_notification.notification_type == NotificationType.EMAIL
     provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-email-tasks", countdown=30
+        [str(persisted_notification.id)], queue="send-email-tasks"
     )
 
 
@@ -822,7 +822,7 @@ def test_should_use_email_template_and_persist_without_personalisation(
     assert not persisted_notification.reference
     assert persisted_notification.notification_type == NotificationType.EMAIL
     provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-email-tasks", countdown=30
+        [str(persisted_notification.id)], queue="send-email-tasks"
     )
 
 
diff --git a/tests/app/organization/test_invite_rest.py b/tests/app/organization/test_invite_rest.py
index 3b3c2387d..23a65dda1 100644
--- a/tests/app/organization/test_invite_rest.py
+++ b/tests/app/organization/test_invite_rest.py
@@ -73,7 +73,7 @@ def test_create_invited_org_user(
     # assert len(notification.personalisation["url"]) > len(expected_start_of_invite_url)
 
     mocked.assert_called_once_with(
-        [(str(notification.id))], queue="notify-internal-tasks"
+        [(str(notification.id))], queue="notify-internal-tasks",  countdown=30
     )
 
 

From 3fba382cfee95c0a55ab627d64327f5ac5fa1a9a Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 14:43:28 -0800
Subject: [PATCH 13/40] fix uuid

---
 .../app/service/send_notification/test_send_notification.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/tests/app/service/send_notification/test_send_notification.py b/tests/app/service/send_notification/test_send_notification.py
index fd37f7592..831803934 100644
--- a/tests/app/service/send_notification/test_send_notification.py
+++ b/tests/app/service/send_notification/test_send_notification.py
@@ -420,7 +420,7 @@ def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker
             response_data = json.loads(response.data)["data"]
             notification_id = response_data["notification"]["id"]
 
-            mocked.assert_called_once_with([notification_id], queue="send-sms-tasks")
+            mocked.assert_called_once_with([notification_id], queue="send-sms-tasks", countdown=30)
             assert response.status_code == 201
             assert notification_id
             assert "subject" not in response_data
@@ -658,7 +658,7 @@ def test_should_send_sms_to_anyone_with_test_key(
         ],
     )
     app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        [fake_uuid], queue="send-sms-tasks"
+        [fake_uuid], queue="send-sms-tasks", countdown=30
     )
     assert response.status_code == 201
 
@@ -735,7 +735,7 @@ def test_should_send_sms_if_team_api_key_and_a_service_user(
     )
 
     app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        [fake_uuid], queue="send-sms-tasks"
+        [fake_uuid], queue="send-sms-tasks", countdown=30
     )
     assert response.status_code == 201
 

From 7794eb29c3883b76ebcae8214c2b4576eedf546f Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 14:51:40 -0800
Subject: [PATCH 14/40] fix uuid

---
 tests/app/celery/test_tasks.py                              | 2 +-
 .../app/service/send_notification/test_send_notification.py | 6 +++---
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/tests/app/celery/test_tasks.py b/tests/app/celery/test_tasks.py
index d4081fc97..12c15d334 100644
--- a/tests/app/celery/test_tasks.py
+++ b/tests/app/celery/test_tasks.py
@@ -707,7 +707,7 @@ def test_save_email_should_use_template_version_from_job_not_latest(
     assert not persisted_notification.sent_by
     assert persisted_notification.notification_type == NotificationType.EMAIL
     provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-email-tasks", countdown=30
+        [str(persisted_notification.id)], queue="send-email-tasks"
     )
 
 
diff --git a/tests/app/service/send_notification/test_send_notification.py b/tests/app/service/send_notification/test_send_notification.py
index 831803934..0ec50428e 100644
--- a/tests/app/service/send_notification/test_send_notification.py
+++ b/tests/app/service/send_notification/test_send_notification.py
@@ -476,7 +476,7 @@ def test_should_allow_valid_email_notification(
             response_data = json.loads(response.get_data(as_text=True))["data"]
             notification_id = response_data["notification"]["id"]
             app.celery.provider_tasks.deliver_email.apply_async.assert_called_once_with(
-                [notification_id], queue="send-email-tasks"
+                [notification_id], queue="send-email-tasks", countdown=30
             )
 
             assert response.status_code == 201
@@ -620,7 +620,7 @@ def test_should_send_email_if_team_api_key_and_a_service_user(
     )
 
     app.celery.provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [fake_uuid], queue="send-email-tasks"
+        [fake_uuid], queue="send-email-tasks", countdown=30
     )
     assert response.status_code == 201
 
@@ -697,7 +697,7 @@ def test_should_send_email_to_anyone_with_test_key(
     )
 
     app.celery.provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [fake_uuid], queue="send-email-tasks"
+        [fake_uuid], queue="send-email-tasks", countdown=30
     )
     assert response.status_code == 201
 

From 2acd0a87b9667423086026f437165caf2bb258b2 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 15:02:43 -0800
Subject: [PATCH 15/40] fix uuid

---
 tests/app/celery/test_tasks.py                             | 7 +++----
 .../service/send_notification/test_send_notification.py    | 4 ++--
 2 files changed, 5 insertions(+), 6 deletions(-)

diff --git a/tests/app/celery/test_tasks.py b/tests/app/celery/test_tasks.py
index 12c15d334..ccc19bc8d 100644
--- a/tests/app/celery/test_tasks.py
+++ b/tests/app/celery/test_tasks.py
@@ -435,7 +435,7 @@ def test_should_send_template_to_correct_sms_task_and_persist(
     assert persisted_notification.personalisation == {}
     assert persisted_notification.notification_type == NotificationType.SMS
     mocked_deliver_sms.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-sms-tasks"
+        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=30
     )
 
 
@@ -471,7 +471,7 @@ def test_should_save_sms_if_restricted_service_and_valid_number(
     assert not persisted_notification.personalisation
     assert persisted_notification.notification_type == NotificationType.SMS
     provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-sms-tasks"
+        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=30
     )
 
 
@@ -670,7 +670,7 @@ def test_should_use_email_template_and_persist(
     assert persisted_notification.notification_type == NotificationType.EMAIL
 
     provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-email-tasks", countdown=30
+        [str(persisted_notification.id)], queue="send-email-tasks"
     )
 
 
@@ -946,7 +946,6 @@ def test_save_sms_uses_sms_sender_reply_to_text(mocker, notify_db_session):
         encryption.encrypt(notification),
     )
 
-    batch_insert_notifications()
     persisted_notification = Notification.query.one()
     assert persisted_notification.reply_to_text == "+12028675309"
 
diff --git a/tests/app/service/send_notification/test_send_notification.py b/tests/app/service/send_notification/test_send_notification.py
index 0ec50428e..d0b49a982 100644
--- a/tests/app/service/send_notification/test_send_notification.py
+++ b/tests/app/service/send_notification/test_send_notification.py
@@ -150,7 +150,7 @@ def test_send_notification_with_placeholders_replaced(
                 {"template_version": sample_email_template_with_placeholders.version}
             )
 
-            mocked.assert_called_once_with([notification_id], queue="send-email-tasks")
+            mocked.assert_called_once_with([notification_id], queue="send-email-tasks", countdown=30)
             assert response.status_code == 201
             assert response_data["body"] == "Hello Jo\nThis is an email from GOV.UK"
             assert response_data["subject"] == "Jo"
@@ -1185,7 +1185,7 @@ def test_should_allow_store_original_number_on_sms_notification(
     response_data = json.loads(response.data)["data"]
     notification_id = response_data["notification"]["id"]
 
-    mocked.assert_called_once_with([notification_id], queue="send-sms-tasks")
+    mocked.assert_called_once_with([notification_id], queue="send-sms-tasks", countdown=30)
     assert response.status_code == 201
     assert notification_id
     notifications = Notification.query.all()

From bf3fc43e87d3898373f26c771ca37d76bfbb3090 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 15:06:35 -0800
Subject: [PATCH 16/40] fix uuid

---
 tests/app/celery/test_tasks.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/tests/app/celery/test_tasks.py b/tests/app/celery/test_tasks.py
index ccc19bc8d..eeff49251 100644
--- a/tests/app/celery/test_tasks.py
+++ b/tests/app/celery/test_tasks.py
@@ -13,7 +13,6 @@ from sqlalchemy.exc import SQLAlchemyError
 
 from app import db, encryption
 from app.celery import provider_tasks, tasks
-from app.celery.scheduled_tasks import batch_insert_notifications
 from app.celery.tasks import (
     get_recipient_csv_and_template_and_sender_id,
     process_incomplete_job,

From 44ce4951900e4bd318f64880bf4f4aa21306762f Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 16:25:26 -0800
Subject: [PATCH 17/40] fix uuid

---
 .../app/notifications/test_process_notification.py |  8 ++++----
 tests/app/organization/test_invite_rest.py         |  2 +-
 .../send_notification/test_send_notification.py    | 14 ++++++++++----
 tests/app/service/test_rest.py                     |  2 +-
 .../app/service_invite/test_service_invite_rest.py |  2 +-
 tests/app/user/test_rest.py                        |  4 ++--
 tests/app/user/test_rest_verify.py                 | 10 ++++++----
 7 files changed, 25 insertions(+), 17 deletions(-)

diff --git a/tests/app/notifications/test_process_notification.py b/tests/app/notifications/test_process_notification.py
index 296f68adf..06314ae75 100644
--- a/tests/app/notifications/test_process_notification.py
+++ b/tests/app/notifications/test_process_notification.py
@@ -263,7 +263,9 @@ def test_send_notification_to_queue(
 
     send_notification_to_queue(notification=notification, queue=requested_queue)
 
-    mocked.assert_called_once_with([str(notification.id)], queue=expected_queue, countdown=30)
+    mocked.assert_called_once_with(
+        [str(notification.id)], queue=expected_queue, countdown=30
+    )
 
 
 def test_send_notification_to_queue_throws_exception_deletes_notification(
@@ -276,9 +278,7 @@ def test_send_notification_to_queue_throws_exception_deletes_notification(
     with pytest.raises(Boto3Error):
         send_notification_to_queue(sample_notification, False)
     mocked.assert_called_once_with(
-        [(str(sample_notification.id))],
-        queue="send-sms-tasks",
-        countdown=30
+        [(str(sample_notification.id))], queue="send-sms-tasks", countdown=30
     )
 
     assert _get_notification_query_count() == 0
diff --git a/tests/app/organization/test_invite_rest.py b/tests/app/organization/test_invite_rest.py
index 23a65dda1..bacab402d 100644
--- a/tests/app/organization/test_invite_rest.py
+++ b/tests/app/organization/test_invite_rest.py
@@ -73,7 +73,7 @@ def test_create_invited_org_user(
     # assert len(notification.personalisation["url"]) > len(expected_start_of_invite_url)
 
     mocked.assert_called_once_with(
-        [(str(notification.id))], queue="notify-internal-tasks",  countdown=30
+        [(str(notification.id))], queue="notify-internal-tasks", countdown=30
     )
 
 
diff --git a/tests/app/service/send_notification/test_send_notification.py b/tests/app/service/send_notification/test_send_notification.py
index d0b49a982..80f14a9c8 100644
--- a/tests/app/service/send_notification/test_send_notification.py
+++ b/tests/app/service/send_notification/test_send_notification.py
@@ -150,7 +150,9 @@ def test_send_notification_with_placeholders_replaced(
                 {"template_version": sample_email_template_with_placeholders.version}
             )
 
-            mocked.assert_called_once_with([notification_id], queue="send-email-tasks", countdown=30)
+            mocked.assert_called_once_with(
+                [notification_id], queue="send-email-tasks", countdown=30
+            )
             assert response.status_code == 201
             assert response_data["body"] == "Hello Jo\nThis is an email from GOV.UK"
             assert response_data["subject"] == "Jo"
@@ -420,7 +422,9 @@ def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker
             response_data = json.loads(response.data)["data"]
             notification_id = response_data["notification"]["id"]
 
-            mocked.assert_called_once_with([notification_id], queue="send-sms-tasks", countdown=30)
+            mocked.assert_called_once_with(
+                [notification_id], queue="send-sms-tasks", countdown=30
+            )
             assert response.status_code == 201
             assert notification_id
             assert "subject" not in response_data
@@ -853,7 +857,7 @@ def test_should_delete_notification_and_return_error_if_redis_fails(
         )
     assert str(e.value) == "failed to talk to redis"
 
-    mocked.assert_called_once_with([fake_uuid], queue=queue_name)
+    mocked.assert_called_once_with([fake_uuid], queue=queue_name, countdown=30)
     assert not notifications_dao.get_notification_by_id(fake_uuid)
     assert not NotificationHistory.query.get(fake_uuid)
 
@@ -1185,7 +1189,9 @@ def test_should_allow_store_original_number_on_sms_notification(
     response_data = json.loads(response.data)["data"]
     notification_id = response_data["notification"]["id"]
 
-    mocked.assert_called_once_with([notification_id], queue="send-sms-tasks", countdown=30)
+    mocked.assert_called_once_with(
+        [notification_id], queue="send-sms-tasks", countdown=30
+    )
     assert response.status_code == 201
     assert notification_id
     notifications = Notification.query.all()
diff --git a/tests/app/service/test_rest.py b/tests/app/service/test_rest.py
index 132de48e9..2b2472ad7 100644
--- a/tests/app/service/test_rest.py
+++ b/tests/app/service/test_rest.py
@@ -3025,7 +3025,7 @@ def test_verify_reply_to_email_address_should_send_verification_email(
     assert notification.template_id == verify_reply_to_address_email_template.id
     assert response["data"] == {"id": str(notification.id)}
     mocked.assert_called_once_with(
-        [str(notification.id)], queue="notify-internal-tasks"
+        [str(notification.id)], queue="notify-internal-tasks", countdown=30
     )
     assert (
         notification.reply_to_text
diff --git a/tests/app/service_invite/test_service_invite_rest.py b/tests/app/service_invite/test_service_invite_rest.py
index 61b8b79e7..c43b2e878 100644
--- a/tests/app/service_invite/test_service_invite_rest.py
+++ b/tests/app/service_invite/test_service_invite_rest.py
@@ -90,7 +90,7 @@ def test_create_invited_user(
     )
 
     mocked.assert_called_once_with(
-        [(str(notification.id))], queue="notify-internal-tasks"
+        [(str(notification.id))], queue="notify-internal-tasks", countdown=30
     )
 
 
diff --git a/tests/app/user/test_rest.py b/tests/app/user/test_rest.py
index f1ea5041b..860e2b10b 100644
--- a/tests/app/user/test_rest.py
+++ b/tests/app/user/test_rest.py
@@ -664,7 +664,7 @@ def test_send_already_registered_email(
     stmt = select(Notification)
     notification = db.session.execute(stmt).scalars().first()
     mocked.assert_called_once_with(
-        ([str(notification.id)]), queue="notify-internal-tasks"
+        ([str(notification.id)]), queue="notify-internal-tasks", countdown=30
     )
     assert (
         notification.reply_to_text
@@ -703,7 +703,7 @@ def test_send_user_confirm_new_email_returns_204(
     stmt = select(Notification)
     notification = db.session.execute(stmt).scalars().first()
     mocked.assert_called_once_with(
-        ([str(notification.id)]), queue="notify-internal-tasks"
+        ([str(notification.id)]), queue="notify-internal-tasks", countdown=30
     )
     assert (
         notification.reply_to_text
diff --git a/tests/app/user/test_rest_verify.py b/tests/app/user/test_rest_verify.py
index d32d923bf..805d90a8e 100644
--- a/tests/app/user/test_rest_verify.py
+++ b/tests/app/user/test_rest_verify.py
@@ -231,7 +231,7 @@ def test_send_user_sms_code(client, sample_user, sms_code_template, mocker):
     assert notification.reply_to_text == notify_service.get_default_sms_sender()
 
     app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        ([str(notification.id)]), queue="notify-internal-tasks"
+        ([str(notification.id)]), queue="notify-internal-tasks", countdown=30
     )
 
 
@@ -267,7 +267,7 @@ def test_send_user_code_for_sms_with_optional_to_field(
     notification = Notification.query.first()
     assert notification.to == "1"
     app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        ([str(notification.id)]), queue="notify-internal-tasks"
+        ([str(notification.id)]), queue="notify-internal-tasks", countdown=30
     )
 
 
@@ -349,7 +349,7 @@ def test_send_new_user_email_verification(
     notification = Notification.query.first()
     assert _get_verify_code_count() == 0
     mocked.assert_called_once_with(
-        ([str(notification.id)]), queue="notify-internal-tasks"
+        ([str(notification.id)]), queue="notify-internal-tasks", countdown=30
     )
     assert (
         notification.reply_to_text
@@ -494,7 +494,9 @@ def test_send_user_email_code(
     )
     assert noti.to == "1"
     assert str(noti.template_id) == current_app.config["EMAIL_2FA_TEMPLATE_ID"]
-    deliver_email.assert_called_once_with([str(noti.id)], queue="notify-internal-tasks")
+    deliver_email.assert_called_once_with(
+        [str(noti.id)], queue="notify-internal-tasks", countdown=30
+    )
 
 
 @pytest.mark.skip(reason="Broken email functionality")

From 28470468e25c84ac689dd43d3fdc9dcf2d18838f Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Fri, 10 Jan 2025 16:38:37 -0800
Subject: [PATCH 18/40] fix uuid

---
 tests/app/service/send_notification/test_send_notification.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/app/service/send_notification/test_send_notification.py b/tests/app/service/send_notification/test_send_notification.py
index 80f14a9c8..dab4ca43f 100644
--- a/tests/app/service/send_notification/test_send_notification.py
+++ b/tests/app/service/send_notification/test_send_notification.py
@@ -796,7 +796,7 @@ def test_should_persist_notification(
         ],
     )
 
-    mocked.assert_called_once_with([fake_uuid], queue=queue_name)
+    mocked.assert_called_once_with([fake_uuid], queue=queue_name, countdown=30)
     assert response.status_code == 201
 
     notification = notifications_dao.get_notification_by_id(fake_uuid)

From f4972037912af5af98fe90f6fa6607180aac12d4 Mon Sep 17 00:00:00 2001
From: Andrew Shumway <ashumway@flexion.us>
Date: Mon, 13 Jan 2025 10:07:33 -0700
Subject: [PATCH 19/40] Ensure created_at stamp is correct

---
 app/dao/notifications_dao.py | 25 ++++++++++++++++++++++++-
 app/service_invite/rest.py   |  8 ++++++--
 2 files changed, 30 insertions(+), 3 deletions(-)

diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py
index 139f7ae8a..691b29065 100644
--- a/app/dao/notifications_dao.py
+++ b/app/dao/notifications_dao.py
@@ -1,5 +1,6 @@
 import json
-from datetime import timedelta
+import os
+from datetime import datetime, timedelta
 from time import time
 
 from flask import current_app
@@ -29,6 +30,7 @@ from app.models import FactNotificationStatus, Notification, NotificationHistory
 from app.utils import (
     escape_special_characters,
     get_midnight_in_utc,
+    hilite,
     midnight_n_days_ago,
     utc_now,
 )
@@ -95,6 +97,27 @@ def dao_create_notification(notification):
     # notify-api-1454 insert only if it doesn't exist
     if not dao_notification_exists(notification.id):
         db.session.add(notification)
+        # There have been issues with invites expiring.
+        # Ensure the created at value is set and debug.
+        if notification.notification_type == "email":
+            orig_time = notification.created_at
+
+            now_time = utc_now()
+            print(hilite(f"original time: {orig_time} - {type(orig_time)} \n now time: {now_time} - {type(now_time)}"))
+            diff_time = now_time - datetime.strptime(orig_time, "%Y-%m-%D-%H-%M-%S")
+            current_app.logger.error(
+                f"dao_create_notification orig created at: {orig_time} and now created at: {now_time}"
+            )
+            if diff_time.total_seconds() > 300:
+                current_app.logger.error(
+                    "Something is wrong with notification.created_at in email!"
+                )
+                if os.getenv("NOTIFY_ENVIRONMENT") not in ["test"]:
+                    notification.created_at = now_time
+                    dao_update_notification(notification)
+                    current_app.logger.error(
+                        f"Email notification created_at reset to   {notification.created_at}"
+                    )
 
 
 def country_records_delivery(phone_prefix):
diff --git a/app/service_invite/rest.py b/app/service_invite/rest.py
index e375b93a5..e1f26236f 100644
--- a/app/service_invite/rest.py
+++ b/app/service_invite/rest.py
@@ -25,7 +25,7 @@ from app.notifications.process_notifications import (
     send_notification_to_queue,
 )
 from app.schemas import invited_user_schema
-from app.utils import utc_now
+from app.utils import hilite, utc_now
 from notifications_utils.url_safe_token import check_token, generate_token
 
 service_invite = Blueprint("service_invite", __name__)
@@ -67,7 +67,7 @@ def _create_service_invite(invited_user, nonce, state):
         "service_name": invited_user.service.name,
         "url": url,
     }
-
+    created_at = utc_now()
     saved_notification = persist_notification(
         template_id=template.id,
         template_version=template.version,
@@ -78,6 +78,10 @@ def _create_service_invite(invited_user, nonce, state):
         api_key_id=None,
         key_type=KeyType.NORMAL,
         reply_to_text=invited_user.from_user.email_address,
+        created_at=created_at,
+    )
+    print(
+        hilite(f"saved notification created at time: {saved_notification.created_at}")
     )
     saved_notification.personalisation = personalisation
     redis_store.set(

From a92eb91470d6e153b8159619aefd63ffa4385611 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 10:00:18 -0800
Subject: [PATCH 20/40] add a test

---
 app/celery/scheduled_tasks.py            | 33 +++++++++++++-----------
 app/dao/notifications_dao.py             | 14 ++++------
 tests/app/celery/test_scheduled_tasks.py | 26 ++++++++++++++++++-
 3 files changed, 48 insertions(+), 25 deletions(-)

diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py
index 9fcfeeb04..12c721114 100644
--- a/app/celery/scheduled_tasks.py
+++ b/app/celery/scheduled_tasks.py
@@ -292,39 +292,42 @@ def cleanup_delivery_receipts(self):
 
 @notify_celery.task(bind=True, name="batch-insert-notifications")
 def batch_insert_notifications(self):
-    current_app.logger.info("ENTER SCHEDULED TASK")
     batch = []
+
+    # TODO We probably need some way to clear the list if
+    # things go haywire.  A command?
+
     # with redis_store.pipeline():
     #     while redis_store.llen("message_queue") > 0:
     #         redis_store.lpop("message_queue")
     #     current_app.logger.info("EMPTY!")
     #     return
+    current_len = redis_store.llen("message_queue")
     with redis_store.pipeline():
-        current_app.logger.info("PIPELINE")
-        # since this list is always growing, just grab what is available when
+        # since this list is being fed by other processes, just grab what is available when
         # this call is made and process that.
-        current_len = redis_store.llen("message_queue")
+
         count = 0
         while count < current_len:
             count = count + 1
             notification_bytes = redis_store.lpop("message_queue")
             notification_dict = json.loads(notification_bytes.decode("utf-8"))
             notification_dict["status"] = notification_dict.pop("notification_status")
-            notification_dict["created_at"] = utc_now()
+            if not notification_dict.get("created_at"):
+                notification_dict["created_at"] = utc_now()
             notification = Notification(**notification_dict)
-            current_app.logger.info(
-                f"WHAT IS THIS NOTIFICATION {type(notification)} {notification}"
-            )
             if notification is not None:
-                current_app.logger.info(
-                    f"SCHEDULED adding notification {notification.id} to batch"
-                )
                 batch.append(notification)
     try:
-        current_app.logger.info("GOING TO DO BATCH INSERT")
         dao_batch_insert_notifications(batch)
     except Exception as e:
         current_app.logger.exception(f"Notification batch insert failed {e}")
-
-        for msg in batch:
-            redis_store.rpush("notification_queue", json.dumps(msg))
+        for n in batch:
+            # Use 'created_at' as a TTL so we don't retry infinitely
+            if n.created_at < utc_now() - timedelta(minutes=1):
+                current_app.logger.warning(
+                    f"Abandoning stale data, could not write to db: {n.serialize_for_redis(n)}"
+                )
+                continue
+            else:
+                redis_store.rpush("message_queue", json.dumps(n.serialize_for_redis(n)))
diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py
index 92dcc234c..fece5b3d2 100644
--- a/app/dao/notifications_dao.py
+++ b/app/dao/notifications_dao.py
@@ -2,7 +2,6 @@ import json
 from datetime import timedelta
 from time import time
 
-import sqlalchemy
 from flask import current_app
 from sqlalchemy import (
     TIMESTAMP,
@@ -803,11 +802,8 @@ def dao_close_out_delivery_receipts():
 
 
 def dao_batch_insert_notifications(batch):
-    current_app.logger.info("DOING BATCH INSERT IN DAO")
-    try:
-        db.session.bulk_save_objects(batch)
-        db.session.commit()
-        current_app.logger.info(f"SUCCESSFULLY INSERTED: {len(batch)}")
-        return len(batch)
-    except sqlalchemy.exc.SQLAlchemyError as e:
-        current_app.logger.exception(f"Error during batch insert {e}")
+
+    db.session.bulk_save_objects(batch)
+    db.session.commit()
+    current_app.logger.info(f"Batch inserted notifications: {len(batch)}")
+    return len(batch)
diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index 0c285ea94..8b5fc6be9 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -1,12 +1,14 @@
+import json
 from collections import namedtuple
 from datetime import timedelta
 from unittest import mock
-from unittest.mock import ANY, call
+from unittest.mock import ANY, MagicMock, call
 
 import pytest
 
 from app.celery import scheduled_tasks
 from app.celery.scheduled_tasks import (
+    batch_insert_notifications,
     check_for_missing_rows_in_completed_jobs,
     check_for_services_with_high_failure_rates_or_sending_to_tv_numbers,
     check_job_status,
@@ -523,3 +525,25 @@ def test_check_for_services_with_high_failure_rates_or_sending_to_tv_numbers(
         technical_ticket=True,
     )
     mock_send_ticket_to_zendesk.assert_called_once()
+
+
+def test_batch_insert_with_valid_notifications(mocker):
+    mocker.patch("app.celery.scheduled_tasks.dao_batch_insert_notifications")
+    rs = MagicMock()
+    mocker.patch("app.celery.scheduled_tasks.redis_store", rs)
+    notifications = [
+        {"id": 1, "notification_status": "pending"},
+        {"id": 2, "notification_status": "pending"},
+    ]
+    serialized_notifications = [json.dumps(n).encode("utf-8") for n in notifications]
+
+    pipeline_mock = MagicMock()
+
+    rs.pipeline.return_value.__enter__.return_value = pipeline_mock
+    rs.llen.return_value = len(notifications)
+    rs.lpop.side_effect = serialized_notifications
+
+    batch_insert_notifications()
+
+    rs.llen.assert_called_once_with("message_queue")
+    rs.lpop.assert_called_with("message_queue")

From 238ec27d4ed12b0ce6413bf87b290707eafd7462 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 10:48:19 -0800
Subject: [PATCH 21/40] more tests

---
 app/celery/scheduled_tasks.py            |  4 +-
 tests/app/celery/test_scheduled_tasks.py | 53 ++++++++++++++++++++++++
 2 files changed, 55 insertions(+), 2 deletions(-)

diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py
index 12c721114..a60551b75 100644
--- a/app/celery/scheduled_tasks.py
+++ b/app/celery/scheduled_tasks.py
@@ -320,8 +320,8 @@ def batch_insert_notifications(self):
                 batch.append(notification)
     try:
         dao_batch_insert_notifications(batch)
-    except Exception as e:
-        current_app.logger.exception(f"Notification batch insert failed {e}")
+    except Exception:
+        current_app.logger.exception("Notification batch insert failed")
         for n in batch:
             # Use 'created_at' as a TTL so we don't retry infinitely
             if n.created_at < utc_now() - timedelta(minutes=1):
diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index 8b5fc6be9..fec64480a 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -547,3 +547,56 @@ def test_batch_insert_with_valid_notifications(mocker):
 
     rs.llen.assert_called_once_with("message_queue")
     rs.lpop.assert_called_with("message_queue")
+
+
+def test_batch_insert_with_expired_notifications(mocker):
+    expired_time = utc_now() - timedelta(minutes=2)
+    mocker.patch(
+        "app.celery.scheduled_tasks.dao_batch_insert_notifications",
+        side_effect=Exception("DB Error"),
+    )
+    rs = MagicMock()
+    mocker.patch("app.celery.scheduled_tasks.redis_store", rs)
+    notifications = [
+        {
+            "id": 1,
+            "notification_status": "pending",
+            "created_at": utc_now().isoformat(),
+        },
+        {
+            "id": 2,
+            "notification_status": "pending",
+            "created_at": expired_time.isoformat(),
+        },
+    ]
+    serialized_notifications = [json.dumps(n).encode("utf-8") for n in notifications]
+
+    pipeline_mock = MagicMock()
+
+    rs.pipeline.return_value.__enter__.return_value = pipeline_mock
+    rs.llen.return_value = len(notifications)
+    rs.lpop.side_effect = serialized_notifications
+
+    batch_insert_notifications()
+
+    rs.llen.assert_called_once_with("message_queue")
+    rs.rpush.assert_called_once()
+    requeued_notification = json.loads(rs.rpush.call_args[0][1])
+    assert requeued_notification["id"] == 1
+
+
+def test_batch_insert_with_malformed_notifications(mocker):
+    rs = MagicMock()
+    mocker.patch("app.celery.scheduled_tasks.redis_store", rs)
+    malformed_data = b"not_a_valid_json"
+    pipeline_mock = MagicMock()
+
+    rs.pipeline.return_value.__enter__.return_value = pipeline_mock
+    rs.llen.return_value = 1
+    rs.lpop.side_effect = [malformed_data]
+
+    with pytest.raises(json.JSONDecodeError):
+        batch_insert_notifications()
+
+    rs.llen.assert_called_once_with("message_queue")
+    rs.rpush.assert_not_called()

From f9641aee39d1885507f9048acf9ba2905afa30ab Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 11:00:42 -0800
Subject: [PATCH 22/40] more tests

---
 app/celery/scheduled_tasks.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py
index a60551b75..ec134c697 100644
--- a/app/celery/scheduled_tasks.py
+++ b/app/celery/scheduled_tasks.py
@@ -1,5 +1,5 @@
 import json
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 from flask import current_app
 from sqlalchemy import between
@@ -324,7 +324,7 @@ def batch_insert_notifications(self):
         current_app.logger.exception("Notification batch insert failed")
         for n in batch:
             # Use 'created_at' as a TTL so we don't retry infinitely
-            if n.created_at < utc_now() - timedelta(minutes=1):
+            if datetime.fromisoformat(n.created_at) < utc_now() - timedelta(minutes=1):
                 current_app.logger.warning(
                     f"Abandoning stale data, could not write to db: {n.serialize_for_redis(n)}"
                 )

From aaddd8c336bf0e03565abfd3317ff9a60aa6a3f6 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 11:10:03 -0800
Subject: [PATCH 23/40] more tests

---
 app/models.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/app/models.py b/app/models.py
index 50c47ec76..f9be291b1 100644
--- a/app/models.py
+++ b/app/models.py
@@ -1702,7 +1702,10 @@ class Notification(db.Model):
                     new_name = "status"
                     value = getattr(obj, new_name)
                 elif column.name == "created_at":
-                    value = (obj.created_at.strftime("%Y-%m-%d %H:%M:%S"),)
+                    if isinstance(obj.created_at, str):
+                        value = obj.created_at
+                    else:
+                        value = (obj.created_at.strftime("%Y-%m-%d %H:%M:%S"),)
                 elif column.name in ["sent_at", "completed_at"]:
                     value = None
                 elif column.name.endswith("_id"):

From 40c3d4a3f2ef6695507334bb489ae555f2f309ca Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 11:37:54 -0800
Subject: [PATCH 24/40] more tests

---
 tests/app/celery/test_scheduled_tasks.py | 25 ++++++++++++++++++++++++
 1 file changed, 25 insertions(+)

diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index fec64480a..30f24b317 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -14,6 +14,7 @@ from app.celery.scheduled_tasks import (
     check_job_status,
     delete_verify_codes,
     expire_or_delete_invitations,
+    process_delivery_receipts,
     replay_created_notifications,
     run_scheduled_jobs,
 )
@@ -600,3 +601,27 @@ def test_batch_insert_with_malformed_notifications(mocker):
 
     rs.llen.assert_called_once_with("message_queue")
     rs.rpush.assert_not_called()
+
+
+def test_process_delivery_receipts_success(mocker):
+    dao_update_mock = mocker.patch(
+        "app.dao.notifications_dao.dao_update_delivery_receipts"
+    )
+    cloudwatch_mock = mocker.patch(
+        "app.clients.cloudwatch.aws_cloudwatch.AwsCloudwatchClient"
+    )
+    cloudwatch_mock.check_delivery_receipts.return_value = {range(2000), range(500)}
+    current_app_mock = mocker.patch("app.celery.scheduled_tasks.current_app")
+    current_app_mock.return_value = MagicMock()
+    processor = MagicMock()
+    processor.process_delivery_receipts = process_delivery_receipts
+
+    processor.process_delivery_receipts()
+
+    cloudwatch_mock.init_app.assert_called_once_with(current_app_mock)
+    cloudwatch_mock.check_delivery_receipts.assert_called_ocne()
+
+    assert dao_update_mock.call_count == 3
+    dao_update_mock.assert_any_call(list(range(1000)), True)
+    dao_update_mock.assert_any_call(list(range(1000, 2000)), True)
+    dao_update_mock.assert_any_call(list(range(500)), True)

From b92430252af4193b2333e93e875a448c07aca196 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 11:47:25 -0800
Subject: [PATCH 25/40] more tests

---
 tests/app/celery/test_scheduled_tasks.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index 30f24b317..161498f27 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -615,6 +615,7 @@ def test_process_delivery_receipts_success(mocker):
     current_app_mock.return_value = MagicMock()
     processor = MagicMock()
     processor.process_delivery_receipts = process_delivery_receipts
+    processor.retry = MagicMock()
 
     processor.process_delivery_receipts()
 
@@ -625,3 +626,4 @@ def test_process_delivery_receipts_success(mocker):
     dao_update_mock.assert_any_call(list(range(1000)), True)
     dao_update_mock.assert_any_call(list(range(1000, 2000)), True)
     dao_update_mock.assert_any_call(list(range(500)), True)
+    processor.retry.assert_not_called()

From af158bf1f0d30e97d60da895929ccfad32d534d1 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 11:59:07 -0800
Subject: [PATCH 26/40] more tests

---
 tests/app/celery/test_scheduled_tasks.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index 161498f27..7964081b7 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -605,10 +605,10 @@ def test_batch_insert_with_malformed_notifications(mocker):
 
 def test_process_delivery_receipts_success(mocker):
     dao_update_mock = mocker.patch(
-        "app.dao.notifications_dao.dao_update_delivery_receipts"
+        "app.celery.scheduled_tasks.dao_update_delivery_receipts"
     )
     cloudwatch_mock = mocker.patch(
-        "app.clients.cloudwatch.aws_cloudwatch.AwsCloudwatchClient"
+        "app.celery.scheduled_tasks.AwsCloudwatchClient"
     )
     cloudwatch_mock.check_delivery_receipts.return_value = {range(2000), range(500)}
     current_app_mock = mocker.patch("app.celery.scheduled_tasks.current_app")
@@ -620,7 +620,7 @@ def test_process_delivery_receipts_success(mocker):
     processor.process_delivery_receipts()
 
     cloudwatch_mock.init_app.assert_called_once_with(current_app_mock)
-    cloudwatch_mock.check_delivery_receipts.assert_called_ocne()
+    cloudwatch_mock.check_delivery_receipts.assert_called_once()
 
     assert dao_update_mock.call_count == 3
     dao_update_mock.assert_any_call(list(range(1000)), True)

From 18debf62e8eed60c03c65f9ef789b5f47f9f2da4 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 12:11:42 -0800
Subject: [PATCH 27/40] more tests

---
 tests/app/celery/test_scheduled_tasks.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index 7964081b7..63038297a 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -610,7 +610,7 @@ def test_process_delivery_receipts_success(mocker):
     cloudwatch_mock = mocker.patch(
         "app.celery.scheduled_tasks.AwsCloudwatchClient"
     )
-    cloudwatch_mock.check_delivery_receipts.return_value = {range(2000), range(500)}
+    cloudwatch_mock.check_delivery_receipts.return_value = (range(2000), range(500))
     current_app_mock = mocker.patch("app.celery.scheduled_tasks.current_app")
     current_app_mock.return_value = MagicMock()
     processor = MagicMock()

From 510b84b96b2a7301d592c105549f416cab9b3cf2 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 12:19:53 -0800
Subject: [PATCH 28/40] more tests

---
 app/celery/scheduled_tasks.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py
index ec134c697..e03545cb8 100644
--- a/app/celery/scheduled_tasks.py
+++ b/app/celery/scheduled_tasks.py
@@ -261,9 +261,11 @@ def process_delivery_receipts(self):
         cloudwatch.init_app(current_app)
         start_time = aware_utcnow() - timedelta(minutes=3)
         end_time = aware_utcnow()
+        print(f"START TIME {start_time} END TIME {end_time}")
         delivered_receipts, failed_receipts = cloudwatch.check_delivery_receipts(
             start_time, end_time
         )
+        print(f"DELIVERED {delivered_receipts} FAILED {failed_receipts}")
         delivered_receipts = list(delivered_receipts)
         for i in range(0, len(delivered_receipts), batch_size):
             batch = delivered_receipts[i : i + batch_size]

From 521ed799e72abd9d45e0bec899e54aa5d8e9af8b Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 12:36:04 -0800
Subject: [PATCH 29/40] more tests

---
 tests/app/celery/test_scheduled_tasks.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index 63038297a..5738faec0 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -610,7 +610,7 @@ def test_process_delivery_receipts_success(mocker):
     cloudwatch_mock = mocker.patch(
         "app.celery.scheduled_tasks.AwsCloudwatchClient"
     )
-    cloudwatch_mock.check_delivery_receipts.return_value = (range(2000), range(500))
+    cloudwatch_mock.return_value.check_delivery_receipts.return_value = (range(2000), range(500))
     current_app_mock = mocker.patch("app.celery.scheduled_tasks.current_app")
     current_app_mock.return_value = MagicMock()
     processor = MagicMock()

From 1ea89ab616304ebc1a452afae1a851e7125e2373 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 12:44:50 -0800
Subject: [PATCH 30/40] more tests

---
 tests/app/celery/test_scheduled_tasks.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index 5738faec0..e2ed00963 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -619,7 +619,6 @@ def test_process_delivery_receipts_success(mocker):
 
     processor.process_delivery_receipts()
 
-    cloudwatch_mock.init_app.assert_called_once_with(current_app_mock)
     cloudwatch_mock.check_delivery_receipts.assert_called_once()
 
     assert dao_update_mock.call_count == 3

From f4b8c040a3792062364f1e90112c47d57943c359 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 13:03:50 -0800
Subject: [PATCH 31/40] more tests

---
 app/celery/scheduled_tasks.py            | 6 ++++++
 tests/app/celery/test_scheduled_tasks.py | 3 ---
 2 files changed, 6 insertions(+), 3 deletions(-)

diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py
index e03545cb8..ab58a3a9f 100644
--- a/app/celery/scheduled_tasks.py
+++ b/app/celery/scheduled_tasks.py
@@ -267,14 +267,20 @@ def process_delivery_receipts(self):
         )
         print(f"DELIVERED {delivered_receipts} FAILED {failed_receipts}")
         delivered_receipts = list(delivered_receipts)
+        print(f"DELIVERED LIST {delivered_receipts}")
         for i in range(0, len(delivered_receipts), batch_size):
             batch = delivered_receipts[i : i + batch_size]
+            print("UPDATING DELIVERY RECEIPTS")
             dao_update_delivery_receipts(batch, True)
+            print("DEIVERY RECEIPTS UPDATED")
         failed_receipts = list(failed_receipts)
         for i in range(0, len(failed_receipts), batch_size):
+            print("UDPATING FAILED RECEIPTS")
             batch = failed_receipts[i : i + batch_size]
             dao_update_delivery_receipts(batch, False)
+            print("FAILED RECEITPS UPDATED")
     except Exception as ex:
+        print(f"EXCEPTION {ex}")
         retry_count = self.request.retries
         wait_time = 3600 * 2**retry_count
         try:
diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index e2ed00963..8e3160ea4 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -618,9 +618,6 @@ def test_process_delivery_receipts_success(mocker):
     processor.retry = MagicMock()
 
     processor.process_delivery_receipts()
-
-    cloudwatch_mock.check_delivery_receipts.assert_called_once()
-
     assert dao_update_mock.call_count == 3
     dao_update_mock.assert_any_call(list(range(1000)), True)
     dao_update_mock.assert_any_call(list(range(1000, 2000)), True)

From 752e5cada9801f0e16815fce7004d1b1eca4d8a4 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 13:12:09 -0800
Subject: [PATCH 32/40] more tests

---
 tests/app/celery/test_scheduled_tasks.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index 8e3160ea4..b2c75bb45 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -621,5 +621,5 @@ def test_process_delivery_receipts_success(mocker):
     assert dao_update_mock.call_count == 3
     dao_update_mock.assert_any_call(list(range(1000)), True)
     dao_update_mock.assert_any_call(list(range(1000, 2000)), True)
-    dao_update_mock.assert_any_call(list(range(500)), True)
+    dao_update_mock.assert_any_call(list(range(500)), False)
     processor.retry.assert_not_called()

From eac21788a16415c522a28995cba85b0f69ee8064 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 13:21:34 -0800
Subject: [PATCH 33/40] clean up

---
 app/celery/scheduled_tasks.py            | 8 --------
 tests/app/celery/test_scheduled_tasks.py | 7 ++++---
 2 files changed, 4 insertions(+), 11 deletions(-)

diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py
index ab58a3a9f..ec134c697 100644
--- a/app/celery/scheduled_tasks.py
+++ b/app/celery/scheduled_tasks.py
@@ -261,26 +261,18 @@ def process_delivery_receipts(self):
         cloudwatch.init_app(current_app)
         start_time = aware_utcnow() - timedelta(minutes=3)
         end_time = aware_utcnow()
-        print(f"START TIME {start_time} END TIME {end_time}")
         delivered_receipts, failed_receipts = cloudwatch.check_delivery_receipts(
             start_time, end_time
         )
-        print(f"DELIVERED {delivered_receipts} FAILED {failed_receipts}")
         delivered_receipts = list(delivered_receipts)
-        print(f"DELIVERED LIST {delivered_receipts}")
         for i in range(0, len(delivered_receipts), batch_size):
             batch = delivered_receipts[i : i + batch_size]
-            print("UPDATING DELIVERY RECEIPTS")
             dao_update_delivery_receipts(batch, True)
-            print("DEIVERY RECEIPTS UPDATED")
         failed_receipts = list(failed_receipts)
         for i in range(0, len(failed_receipts), batch_size):
-            print("UDPATING FAILED RECEIPTS")
             batch = failed_receipts[i : i + batch_size]
             dao_update_delivery_receipts(batch, False)
-            print("FAILED RECEITPS UPDATED")
     except Exception as ex:
-        print(f"EXCEPTION {ex}")
         retry_count = self.request.retries
         wait_time = 3600 * 2**retry_count
         try:
diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index b2c75bb45..faee04081 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -607,10 +607,11 @@ def test_process_delivery_receipts_success(mocker):
     dao_update_mock = mocker.patch(
         "app.celery.scheduled_tasks.dao_update_delivery_receipts"
     )
-    cloudwatch_mock = mocker.patch(
-        "app.celery.scheduled_tasks.AwsCloudwatchClient"
+    cloudwatch_mock = mocker.patch("app.celery.scheduled_tasks.AwsCloudwatchClient")
+    cloudwatch_mock.return_value.check_delivery_receipts.return_value = (
+        range(2000),
+        range(500),
     )
-    cloudwatch_mock.return_value.check_delivery_receipts.return_value = (range(2000), range(500))
     current_app_mock = mocker.patch("app.celery.scheduled_tasks.current_app")
     current_app_mock.return_value = MagicMock()
     processor = MagicMock()

From 4965bc2354dcdf73bbf4667ebc7952cb4eccc66a Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 13:35:40 -0800
Subject: [PATCH 34/40] change countdown from 30 to 60 seconds for message
 sends to better match batch insert timing

---
 app/celery/scheduled_tasks.py                 |  2 +-
 app/celery/tasks.py                           |  2 +-
 app/notifications/process_notifications.py    |  2 +-
 tests/app/celery/test_scheduled_tasks.py      |  4 ++--
 tests/app/celery/test_tasks.py                |  6 +++---
 .../test_process_notification.py              |  4 ++--
 tests/app/organization/test_invite_rest.py    |  2 +-
 .../test_send_notification.py                 | 20 +++++++++----------
 tests/app/service/test_rest.py                |  2 +-
 .../test_service_invite_rest.py               |  2 +-
 tests/app/user/test_rest.py                   |  4 ++--
 tests/app/user/test_rest_verify.py            |  8 ++++----
 12 files changed, 29 insertions(+), 29 deletions(-)

diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py
index ec134c697..a7fe15b75 100644
--- a/app/celery/scheduled_tasks.py
+++ b/app/celery/scheduled_tasks.py
@@ -324,7 +324,7 @@ def batch_insert_notifications(self):
         current_app.logger.exception("Notification batch insert failed")
         for n in batch:
             # Use 'created_at' as a TTL so we don't retry infinitely
-            if datetime.fromisoformat(n.created_at) < utc_now() - timedelta(minutes=1):
+            if datetime.fromisoformat(n.created_at) < utc_now() - timedelta(seconds=50):
                 current_app.logger.warning(
                     f"Abandoning stale data, could not write to db: {n.serialize_for_redis(n)}"
                 )
diff --git a/app/celery/tasks.py b/app/celery/tasks.py
index 4086f684a..331d95364 100644
--- a/app/celery/tasks.py
+++ b/app/celery/tasks.py
@@ -256,7 +256,7 @@ def save_sms(self, service_id, notification_id, encrypted_notification, sender_i
             )
         )
         provider_tasks.deliver_sms.apply_async(
-            [str(saved_notification.id)], queue=QueueNames.SEND_SMS, countdown=30
+            [str(saved_notification.id)], queue=QueueNames.SEND_SMS, countdown=60
         )
 
         current_app.logger.debug(
diff --git a/app/notifications/process_notifications.py b/app/notifications/process_notifications.py
index 02eb1f766..6b78ce753 100644
--- a/app/notifications/process_notifications.py
+++ b/app/notifications/process_notifications.py
@@ -174,7 +174,7 @@ def send_notification_to_queue_detached(
         deliver_task = provider_tasks.deliver_email
 
     try:
-        deliver_task.apply_async([str(notification_id)], queue=queue, countdown=30)
+        deliver_task.apply_async([str(notification_id)], queue=queue, countdown=60)
     except Exception:
         dao_delete_notifications_by_id(notification_id)
         raise
diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py
index faee04081..76395832e 100644
--- a/tests/app/celery/test_scheduled_tasks.py
+++ b/tests/app/celery/test_scheduled_tasks.py
@@ -311,10 +311,10 @@ def test_replay_created_notifications(notify_db_session, sample_service, mocker)
 
     replay_created_notifications()
     email_delivery_queue.assert_called_once_with(
-        [str(old_email.id)], queue="send-email-tasks", countdown=30
+        [str(old_email.id)], queue="send-email-tasks", countdown=60
     )
     sms_delivery_queue.assert_called_once_with(
-        [str(old_sms.id)], queue="send-sms-tasks", countdown=30
+        [str(old_sms.id)], queue="send-sms-tasks", countdown=60
     )
 
 
diff --git a/tests/app/celery/test_tasks.py b/tests/app/celery/test_tasks.py
index eeff49251..631b02a78 100644
--- a/tests/app/celery/test_tasks.py
+++ b/tests/app/celery/test_tasks.py
@@ -434,7 +434,7 @@ def test_should_send_template_to_correct_sms_task_and_persist(
     assert persisted_notification.personalisation == {}
     assert persisted_notification.notification_type == NotificationType.SMS
     mocked_deliver_sms.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=30
+        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=60
     )
 
 
@@ -470,7 +470,7 @@ def test_should_save_sms_if_restricted_service_and_valid_number(
     assert not persisted_notification.personalisation
     assert persisted_notification.notification_type == NotificationType.SMS
     provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=30
+        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=60
     )
 
 
@@ -598,7 +598,7 @@ def test_should_save_sms_template_to_and_persist_with_job_id(sample_job, mocker)
     assert persisted_notification.notification_type == NotificationType.SMS
 
     provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=30
+        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=60
     )
 
 
diff --git a/tests/app/notifications/test_process_notification.py b/tests/app/notifications/test_process_notification.py
index 06314ae75..84df3ac05 100644
--- a/tests/app/notifications/test_process_notification.py
+++ b/tests/app/notifications/test_process_notification.py
@@ -264,7 +264,7 @@ def test_send_notification_to_queue(
     send_notification_to_queue(notification=notification, queue=requested_queue)
 
     mocked.assert_called_once_with(
-        [str(notification.id)], queue=expected_queue, countdown=30
+        [str(notification.id)], queue=expected_queue, countdown=60
     )
 
 
@@ -278,7 +278,7 @@ def test_send_notification_to_queue_throws_exception_deletes_notification(
     with pytest.raises(Boto3Error):
         send_notification_to_queue(sample_notification, False)
     mocked.assert_called_once_with(
-        [(str(sample_notification.id))], queue="send-sms-tasks", countdown=30
+        [(str(sample_notification.id))], queue="send-sms-tasks", countdown=60
     )
 
     assert _get_notification_query_count() == 0
diff --git a/tests/app/organization/test_invite_rest.py b/tests/app/organization/test_invite_rest.py
index bacab402d..67d80b8cd 100644
--- a/tests/app/organization/test_invite_rest.py
+++ b/tests/app/organization/test_invite_rest.py
@@ -73,7 +73,7 @@ def test_create_invited_org_user(
     # assert len(notification.personalisation["url"]) > len(expected_start_of_invite_url)
 
     mocked.assert_called_once_with(
-        [(str(notification.id))], queue="notify-internal-tasks", countdown=30
+        [(str(notification.id))], queue="notify-internal-tasks", countdown=60
     )
 
 
diff --git a/tests/app/service/send_notification/test_send_notification.py b/tests/app/service/send_notification/test_send_notification.py
index dab4ca43f..32d4c9ab9 100644
--- a/tests/app/service/send_notification/test_send_notification.py
+++ b/tests/app/service/send_notification/test_send_notification.py
@@ -151,7 +151,7 @@ def test_send_notification_with_placeholders_replaced(
             )
 
             mocked.assert_called_once_with(
-                [notification_id], queue="send-email-tasks", countdown=30
+                [notification_id], queue="send-email-tasks", countdown=60
             )
             assert response.status_code == 201
             assert response_data["body"] == "Hello Jo\nThis is an email from GOV.UK"
@@ -423,7 +423,7 @@ def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker
             notification_id = response_data["notification"]["id"]
 
             mocked.assert_called_once_with(
-                [notification_id], queue="send-sms-tasks", countdown=30
+                [notification_id], queue="send-sms-tasks", countdown=60
             )
             assert response.status_code == 201
             assert notification_id
@@ -480,7 +480,7 @@ def test_should_allow_valid_email_notification(
             response_data = json.loads(response.get_data(as_text=True))["data"]
             notification_id = response_data["notification"]["id"]
             app.celery.provider_tasks.deliver_email.apply_async.assert_called_once_with(
-                [notification_id], queue="send-email-tasks", countdown=30
+                [notification_id], queue="send-email-tasks", countdown=60
             )
 
             assert response.status_code == 201
@@ -624,7 +624,7 @@ def test_should_send_email_if_team_api_key_and_a_service_user(
     )
 
     app.celery.provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [fake_uuid], queue="send-email-tasks", countdown=30
+        [fake_uuid], queue="send-email-tasks", countdown=60
     )
     assert response.status_code == 201
 
@@ -662,7 +662,7 @@ def test_should_send_sms_to_anyone_with_test_key(
         ],
     )
     app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        [fake_uuid], queue="send-sms-tasks", countdown=30
+        [fake_uuid], queue="send-sms-tasks", countdown=60
     )
     assert response.status_code == 201
 
@@ -701,7 +701,7 @@ def test_should_send_email_to_anyone_with_test_key(
     )
 
     app.celery.provider_tasks.deliver_email.apply_async.assert_called_once_with(
-        [fake_uuid], queue="send-email-tasks", countdown=30
+        [fake_uuid], queue="send-email-tasks", countdown=60
     )
     assert response.status_code == 201
 
@@ -739,7 +739,7 @@ def test_should_send_sms_if_team_api_key_and_a_service_user(
     )
 
     app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        [fake_uuid], queue="send-sms-tasks", countdown=30
+        [fake_uuid], queue="send-sms-tasks", countdown=60
     )
     assert response.status_code == 201
 
@@ -796,7 +796,7 @@ def test_should_persist_notification(
         ],
     )
 
-    mocked.assert_called_once_with([fake_uuid], queue=queue_name, countdown=30)
+    mocked.assert_called_once_with([fake_uuid], queue=queue_name, countdown=60)
     assert response.status_code == 201
 
     notification = notifications_dao.get_notification_by_id(fake_uuid)
@@ -857,7 +857,7 @@ def test_should_delete_notification_and_return_error_if_redis_fails(
         )
     assert str(e.value) == "failed to talk to redis"
 
-    mocked.assert_called_once_with([fake_uuid], queue=queue_name, countdown=30)
+    mocked.assert_called_once_with([fake_uuid], queue=queue_name, countdown=60)
     assert not notifications_dao.get_notification_by_id(fake_uuid)
     assert not NotificationHistory.query.get(fake_uuid)
 
@@ -1190,7 +1190,7 @@ def test_should_allow_store_original_number_on_sms_notification(
     notification_id = response_data["notification"]["id"]
 
     mocked.assert_called_once_with(
-        [notification_id], queue="send-sms-tasks", countdown=30
+        [notification_id], queue="send-sms-tasks", countdown=60
     )
     assert response.status_code == 201
     assert notification_id
diff --git a/tests/app/service/test_rest.py b/tests/app/service/test_rest.py
index 2b2472ad7..7efac478a 100644
--- a/tests/app/service/test_rest.py
+++ b/tests/app/service/test_rest.py
@@ -3025,7 +3025,7 @@ def test_verify_reply_to_email_address_should_send_verification_email(
     assert notification.template_id == verify_reply_to_address_email_template.id
     assert response["data"] == {"id": str(notification.id)}
     mocked.assert_called_once_with(
-        [str(notification.id)], queue="notify-internal-tasks", countdown=30
+        [str(notification.id)], queue="notify-internal-tasks", countdown=60
     )
     assert (
         notification.reply_to_text
diff --git a/tests/app/service_invite/test_service_invite_rest.py b/tests/app/service_invite/test_service_invite_rest.py
index c43b2e878..431bb4b8c 100644
--- a/tests/app/service_invite/test_service_invite_rest.py
+++ b/tests/app/service_invite/test_service_invite_rest.py
@@ -90,7 +90,7 @@ def test_create_invited_user(
     )
 
     mocked.assert_called_once_with(
-        [(str(notification.id))], queue="notify-internal-tasks", countdown=30
+        [(str(notification.id))], queue="notify-internal-tasks", countdown=60
     )
 
 
diff --git a/tests/app/user/test_rest.py b/tests/app/user/test_rest.py
index 860e2b10b..171e88d38 100644
--- a/tests/app/user/test_rest.py
+++ b/tests/app/user/test_rest.py
@@ -664,7 +664,7 @@ def test_send_already_registered_email(
     stmt = select(Notification)
     notification = db.session.execute(stmt).scalars().first()
     mocked.assert_called_once_with(
-        ([str(notification.id)]), queue="notify-internal-tasks", countdown=30
+        ([str(notification.id)]), queue="notify-internal-tasks", countdown=60
     )
     assert (
         notification.reply_to_text
@@ -703,7 +703,7 @@ def test_send_user_confirm_new_email_returns_204(
     stmt = select(Notification)
     notification = db.session.execute(stmt).scalars().first()
     mocked.assert_called_once_with(
-        ([str(notification.id)]), queue="notify-internal-tasks", countdown=30
+        ([str(notification.id)]), queue="notify-internal-tasks", countdown=60
     )
     assert (
         notification.reply_to_text
diff --git a/tests/app/user/test_rest_verify.py b/tests/app/user/test_rest_verify.py
index 805d90a8e..64a07d422 100644
--- a/tests/app/user/test_rest_verify.py
+++ b/tests/app/user/test_rest_verify.py
@@ -231,7 +231,7 @@ def test_send_user_sms_code(client, sample_user, sms_code_template, mocker):
     assert notification.reply_to_text == notify_service.get_default_sms_sender()
 
     app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        ([str(notification.id)]), queue="notify-internal-tasks", countdown=30
+        ([str(notification.id)]), queue="notify-internal-tasks", countdown=60
     )
 
 
@@ -267,7 +267,7 @@ def test_send_user_code_for_sms_with_optional_to_field(
     notification = Notification.query.first()
     assert notification.to == "1"
     app.celery.provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        ([str(notification.id)]), queue="notify-internal-tasks", countdown=30
+        ([str(notification.id)]), queue="notify-internal-tasks", countdown=60
     )
 
 
@@ -349,7 +349,7 @@ def test_send_new_user_email_verification(
     notification = Notification.query.first()
     assert _get_verify_code_count() == 0
     mocked.assert_called_once_with(
-        ([str(notification.id)]), queue="notify-internal-tasks", countdown=30
+        ([str(notification.id)]), queue="notify-internal-tasks", countdown=60
     )
     assert (
         notification.reply_to_text
@@ -495,7 +495,7 @@ def test_send_user_email_code(
     assert noti.to == "1"
     assert str(noti.template_id) == current_app.config["EMAIL_2FA_TEMPLATE_ID"]
     deliver_email.assert_called_once_with(
-        [str(noti.id)], queue="notify-internal-tasks", countdown=30
+        [str(noti.id)], queue="notify-internal-tasks", countdown=60
     )
 
 

From ba4301fc4629023b51aedef45b27ad448a6b16f5 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Mon, 13 Jan 2025 14:21:43 -0800
Subject: [PATCH 35/40] fix bug with created_at

---
 app/celery/scheduled_tasks.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py
index a7fe15b75..e3daa0201 100644
--- a/app/celery/scheduled_tasks.py
+++ b/app/celery/scheduled_tasks.py
@@ -315,6 +315,8 @@ def batch_insert_notifications(self):
             notification_dict["status"] = notification_dict.pop("notification_status")
             if not notification_dict.get("created_at"):
                 notification_dict["created_at"] = utc_now()
+            elif isinstance(notification_dict["created_at"], list):
+                notification_dict["created_at"] = notification_dict["created_at"][0]
             notification = Notification(**notification_dict)
             if notification is not None:
                 batch.append(notification)
@@ -324,7 +326,10 @@ def batch_insert_notifications(self):
         current_app.logger.exception("Notification batch insert failed")
         for n in batch:
             # Use 'created_at' as a TTL so we don't retry infinitely
-            if datetime.fromisoformat(n.created_at) < utc_now() - timedelta(seconds=50):
+            notification_time = n.created_at
+            if isinstance(notification_time, str):
+                notification_time = datetime.fromisoformat(n.created_at)
+            if notification_time < utc_now() - timedelta(seconds=50):
                 current_app.logger.warning(
                     f"Abandoning stale data, could not write to db: {n.serialize_for_redis(n)}"
                 )

From 59dfb05ee5dcb850ea1293c54d43693612eadfc0 Mon Sep 17 00:00:00 2001
From: Kenneth Kehl <@kkehl@flexion.us>
Date: Tue, 14 Jan 2025 07:35:02 -0800
Subject: [PATCH 36/40] code review feedback

---
 app/commands.py                                   | 11 +++++++++++
 app/models.py                                     |  2 --
 notifications_utils/clients/redis/redis_client.py |  9 ++++++---
 3 files changed, 17 insertions(+), 5 deletions(-)

diff --git a/app/commands.py b/app/commands.py
index 79bd3192d..58bd542eb 100644
--- a/app/commands.py
+++ b/app/commands.py
@@ -789,6 +789,17 @@ def _update_template(id, name, template_type, content, subject):
     db.session.commit()
 
 
+@notify_command(name="clear-redis-list")
+@click.option("-n", "--name_of_list", required=True)
+def clear_redis_list(name_of_list):
+    my_len_before = redis_store.llen(name_of_list)
+    redis_store.ltrim(name_of_list, 1, 0)
+    my_len_after = redis_store.llen(name_of_list)
+    current_app.logger.info(
+        f"Cleared redis list {name_of_list}. Before: {my_len_before} after {my_len_after}"
+    )
+
+
 @notify_command(name="update-templates")
 def update_templates():
     with open(current_app.config["CONFIG_FILES"] + "/templates.json") as f:
diff --git a/app/models.py b/app/models.py
index f9be291b1..f78f630ea 100644
--- a/app/models.py
+++ b/app/models.py
@@ -1717,8 +1717,6 @@ class Notification(db.Model):
                     pass  # do nothing because we don't have the message id yet
                 else:
                     fields[column.name] = value
-            current_app.logger.warning(f"FIELDS {fields}")
-            print(f"FIELDS {fields}", flush=True)
 
             return fields
         raise ValueError("Provided object is not a SQLAlchemy instance")
diff --git a/notifications_utils/clients/redis/redis_client.py b/notifications_utils/clients/redis/redis_client.py
index c41318243..d96f967a2 100644
--- a/notifications_utils/clients/redis/redis_client.py
+++ b/notifications_utils/clients/redis/redis_client.py
@@ -38,9 +38,8 @@ class RedisClient:
     active = False
     scripts = {}
 
-    @classmethod
-    def pipeline(cls):
-        return cls.redis_store.pipeline()
+    def pipeline(self):
+        return self.redis_store.pipeline()
 
     def init_app(self, app):
         self.active = app.config.get("REDIS_ENABLED")
@@ -172,6 +171,10 @@ class RedisClient:
         if self.active:
             return self.redis_store.llen(key)
 
+    def ltrim(self, key, start, end):
+        if self.active:
+            return self.redis_store.ltrim(key, start, end)
+
     def delete(self, *keys, raise_exception=False):
         keys = [prepare_value(k) for k in keys]
         if self.active:

From 3fd8009e3336e5522e35492e9ff0c34ed8ea8911 Mon Sep 17 00:00:00 2001
From: Andrew Shumway <ashumway@flexion.us>
Date: Wed, 15 Jan 2025 12:49:47 -0700
Subject: [PATCH 37/40] Add error handling for possible string/datetime created
 at stamps

---
 app/dao/notifications_dao.py             | 15 +++++++++++++--
 tests/app/celery/test_reporting_tasks.py |  1 -
 2 files changed, 13 insertions(+), 3 deletions(-)

diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py
index 4ad50c111..b5690e535 100644
--- a/app/dao/notifications_dao.py
+++ b/app/dao/notifications_dao.py
@@ -103,8 +103,19 @@ def dao_create_notification(notification):
             orig_time = notification.created_at
 
             now_time = utc_now()
-            print(hilite(f"original time: {orig_time} - {type(orig_time)} \n now time: {now_time} - {type(now_time)}"))
-            diff_time = now_time - datetime.strptime(orig_time, "%Y-%m-%D-%H-%M-%S")
+            print(
+                hilite(
+                    f"original time: {orig_time} - {type(orig_time)} \n now time: {now_time} - {type(now_time)}"
+                )
+            )
+            try:
+                diff_time = now_time - orig_time
+            except TypeError:
+                try:
+                    orig_time = datetime.strptime(orig_time, "%Y-%m-%dT%H:%M:%S.%fZ")
+                except ValueError:
+                    orig_time = datetime.strptime(orig_time, "%Y-%m-%d")
+                diff_time = now_time - orig_time
             current_app.logger.error(
                 f"dao_create_notification orig created at: {orig_time} and now created at: {now_time}"
             )
diff --git a/tests/app/celery/test_reporting_tasks.py b/tests/app/celery/test_reporting_tasks.py
index 124038d48..952c65e09 100644
--- a/tests/app/celery/test_reporting_tasks.py
+++ b/tests/app/celery/test_reporting_tasks.py
@@ -103,7 +103,6 @@ def test_create_nightly_notification_status_triggers_relevant_tasks(
     mock_celery = mocker.patch(
         "app.celery.reporting_tasks.create_nightly_notification_status_for_service_and_day"
     ).apply_async
-
     for notification_type in NotificationType:
         template = create_template(sample_service, template_type=notification_type)
         create_notification(template=template, created_at=notification_date)

From f1118d6a198b5279011679e6e383206b77ba7867 Mon Sep 17 00:00:00 2001
From: Andrew Shumway <ashumway@flexion.us>
Date: Thu, 16 Jan 2025 08:59:27 -0700
Subject: [PATCH 38/40] Remove print statement

---
 app/dao/notifications_dao.py | 6 ------
 1 file changed, 6 deletions(-)

diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py
index b5690e535..ba04f24ba 100644
--- a/app/dao/notifications_dao.py
+++ b/app/dao/notifications_dao.py
@@ -101,13 +101,7 @@ def dao_create_notification(notification):
         # Ensure the created at value is set and debug.
         if notification.notification_type == "email":
             orig_time = notification.created_at
-
             now_time = utc_now()
-            print(
-                hilite(
-                    f"original time: {orig_time} - {type(orig_time)} \n now time: {now_time} - {type(now_time)}"
-                )
-            )
             try:
                 diff_time = now_time - orig_time
             except TypeError:

From 7a7daf8323724a07e922f6b775e50defe5774b97 Mon Sep 17 00:00:00 2001
From: Andrew Shumway <ashumway@flexion.us>
Date: Thu, 16 Jan 2025 09:02:07 -0700
Subject: [PATCH 39/40] Remove another print statement

---
 app/service_invite/rest.py | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/app/service_invite/rest.py b/app/service_invite/rest.py
index e1f26236f..d59af35ca 100644
--- a/app/service_invite/rest.py
+++ b/app/service_invite/rest.py
@@ -80,9 +80,6 @@ def _create_service_invite(invited_user, nonce, state):
         reply_to_text=invited_user.from_user.email_address,
         created_at=created_at,
     )
-    print(
-        hilite(f"saved notification created at time: {saved_notification.created_at}")
-    )
     saved_notification.personalisation = personalisation
     redis_store.set(
         f"email-personalisation-{saved_notification.id}",

From d7c97d64280b07e31d1d1d121b2fd1f75f7ff4fa Mon Sep 17 00:00:00 2001
From: Andrew Shumway <ashumway@flexion.us>
Date: Thu, 16 Jan 2025 09:05:49 -0700
Subject: [PATCH 40/40] Remove hilite imports

---
 app/dao/notifications_dao.py | 1 -
 app/service_invite/rest.py   | 2 +-
 2 files changed, 1 insertion(+), 2 deletions(-)

diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py
index e1aed5037..fed5d1be8 100644
--- a/app/dao/notifications_dao.py
+++ b/app/dao/notifications_dao.py
@@ -30,7 +30,6 @@ from app.models import FactNotificationStatus, Notification, NotificationHistory
 from app.utils import (
     escape_special_characters,
     get_midnight_in_utc,
-    hilite,
     midnight_n_days_ago,
     utc_now,
 )
diff --git a/app/service_invite/rest.py b/app/service_invite/rest.py
index d59af35ca..f53556b95 100644
--- a/app/service_invite/rest.py
+++ b/app/service_invite/rest.py
@@ -25,7 +25,7 @@ from app.notifications.process_notifications import (
     send_notification_to_queue,
 )
 from app.schemas import invited_user_schema
-from app.utils import hilite, utc_now
+from app.utils import utc_now
 from notifications_utils.url_safe_token import check_token, generate_token
 
 service_invite = Blueprint("service_invite", __name__)