Dehydrate timeout_notifications nightly task

app/dao/notifications_dao.py

@@ -15,7 +15,7 @@
     convert_local_timezone_to_utc,
     convert_utc_to_local_timezone,
 )
-from sqlalchemy import asc, desc, func
+from sqlalchemy import asc, desc, func, update
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.orm import defer, joinedload
 from sqlalchemy.orm.exc import NoResultFound
@@ -48,7 +48,10 @@
     ScheduledNotification,
     Service,
     ServiceDataRetention,
+    Template,
+    compute_formatted_status,
 )
+from app.types import NotificationCallbackData
 from app.utils import escape_special_characters
 
 
@@ -238,24 +241,24 @@ def _update_notification_status(
 
 @transactional
 def _update_notification_statuses(updates):
-    for update in updates:
-        notification = update.get("notification")
-        bounce_response = update.get("bounce_response")
-        provider_response = update.get("provider_response")
-        feedback_reason = update.get("feedback_reason")
+    for u in updates:
+        notification = u.get("notification")
+        bounce_response = u.get("bounce_response")
+        provider_response = u.get("provider_response")
+        feedback_reason = u.get("feedback_reason")
 
-        final_status = _decide_permanent_temporary_failure(current_status=notification.status, status=update.get("new_status"))
+        final_status = _decide_permanent_temporary_failure(current_status=notification.status, status=u.get("new_status"))
         notification.status = final_status
         if provider_response:
-            notification.provider_response = update.get("provider_response")
+            notification.provider_response = u.get("provider_response")
         if bounce_response:
             notification.feedback_type = bounce_response.get("feedback_type")
             notification.feedback_subtype = bounce_response.get("feedback_subtype")
             notification.ses_feedback_id = bounce_response.get("ses_feedback_id")
             notification.ses_feedback_date = bounce_response.get("ses_feedback_date")
         if feedback_reason:
             notification.feedback_reason = feedback_reason
-    update_notification_statuses([update.get("notification") for update in updates])
+    update_notification_statuses([u.get("notification") for u in updates])
 
 
 @transactional
@@ -554,17 +557,63 @@ def dao_delete_notifications_by_id(notification_id):
 
 
 def _timeout_notifications(current_statuses, new_status, timeout_start, updated_at):
-    notifications = Notification.query.filter(
-        Notification.created_at < timeout_start,
-        Notification.status.in_(current_statuses),
-        Notification.notification_type != LETTER_TYPE,
-    ).all()
-    Notification.query.filter(
-        Notification.created_at < timeout_start,
-        Notification.status.in_(current_statuses),
-        Notification.notification_type != LETTER_TYPE,
-    ).update({"status": new_status, "updated_at": updated_at}, synchronize_session=False)
-    return notifications
+    # Uses Postgres UPDATE … RETURNING as a CTE so the bulk status transition
+    # and the callback data fetch happen in a single atomic round-trip.
+    # This avoids materialising a potentially large ID list in Python
+    n = Notification.__table__
+    t = Template.__table__
+
+    update_cte = (
+        update(n)
+        .where(
+            n.c.created_at < timeout_start,
+            n.c.status.in_(current_statuses),
+            n.c.notification_type != LETTER_TYPE,
+        )
+        .values({n.c.status: new_status, n.c.updated_at: updated_at})
+        .returning(
+            n.c.id,
+            n.c.service_id,
+            n.c["to"],
+            n.c.notification_type,
+            n.c.client_reference,
+            n.c.provider_response,
+            n.c.updated_at,
+            n.c.created_at,
+            n.c.sent_at,
+            n.c.feedback_subtype,
+            n.c.feedback_reason,
+            n.c.template_id,
+        )
+        .cte("updated")
+    )
+
+    stmt = update_cte.join(t, t.c.id == update_cte.c.template_id).select()
+
+    rows = db.session.execute(stmt).fetchall()
+
+    # Lightweight dataclass with pre-computed formatted_status
+    return [
+        NotificationCallbackData(
+            id=str(row.id),
+            service_id=str(row.service_id),
+            to=row._mapping["to"],
+            status=new_status,
+            formatted_status=compute_formatted_status(
+                row.template_type,
+                new_status,
+                row.feedback_subtype,
+                row.feedback_reason,
+            ),
+            notification_type=row.notification_type,
+            client_reference=row.client_reference,
+            provider_response=row.provider_response,
+            created_at=row.created_at,
+            updated_at=row.updated_at,
+            sent_at=row.sent_at,
+        )
+        for row in rows
+    ]
 
 
 def dao_timeout_notifications(timeout_period_in_seconds):

app/models.py

@@ -2,7 +2,7 @@
 import itertools
 import uuid
 from enum import Enum, StrEnum
-from typing import Any, Iterable, Literal
+from typing import Any, Iterable, Literal, Optional
 
 from flask import current_app, url_for
 from flask_sqlalchemy.model import DefaultMeta
@@ -1773,6 +1773,53 @@ def check_code(self, cde):
 NOTIFICATION_UNKNOWN_BOUNCE_SUBTYPE = "unknown-bounce-subtype"
 
 
+def compute_formatted_status(
+    template_type: str,
+    status: str,
+    feedback_subtype: Optional[str] = None,
+    feedback_reason: Optional[str] = None,
+) -> str:
+    """Compute the human-readable notification status string from scalar values.
+
+    Accepts plain scalars so callers (e.g. _timeout_notifications) do not need to hydrate a full ORM object.
+    """
+    permanent_failure_status = (
+        {
+            "suppressed": "Blocked",
+            "on-account-suppression-list": "Blocked",
+        }.get(feedback_subtype, "No such address")
+        if feedback_subtype
+        else "No such address"
+    )
+
+    provider_failure_status = (
+        {
+            "NO_ORIGINATION_IDENTITIES_FOUND": "Can't send to this international number",
+            "DESTINATION_COUNTRY_BLOCKED": "Can't send to this international number",
+        }.get(feedback_reason, "No such number")
+        if feedback_reason
+        else "No such number"
+    )
+
+    return {
+        "email": {
+            **EMAIL_STATUS_FORMATTED,
+            "permanent-failure": permanent_failure_status,
+        },
+        "sms": {
+            **SMS_STATUS_FORMATTED,
+            "provider-failure": provider_failure_status,
+        },
+        "letter": {
+            "technical-failure": "Technical failure",
+            "sending": "Accepted",
+            "created": "Accepted",
+            "delivered": "Received",
+            "returned-letter": "Returned",
+        },
+    }[template_type].get(status, status)
+
+
 class NotificationStatusTypes(BaseModel):
     __tablename__ = "notification_status_types"
 
@@ -1965,45 +2012,12 @@ def subject(self):
 
     @property
     def formatted_status(self):
-        def _getStatusByBounceSubtype():
-            """Return the status of a notification based on the bounce sub type"""
-            # note: if this function changes, update the report query in app/report/utils.py::build_notifications_query
-            if self.feedback_subtype:
-                return {
-                    "suppressed": "Blocked",
-                    "on-account-suppression-list": "Blocked",
-                }.get(self.feedback_subtype, "No such address")
-            else:
-                return "No such address"
-
-        def _get_sms_status_by_feedback_reason():
-            """Return the status of a notification based on the feedback reason"""
-            # note: if this function changes, update the report query in app/report/utils.py::build_notifications_query
-            if self.feedback_reason:
-                return {
-                    "NO_ORIGINATION_IDENTITIES_FOUND": "Can't send to this international number",
-                    "DESTINATION_COUNTRY_BLOCKED": "Can't send to this international number",
-                }.get(self.feedback_reason, "No such number")
-            else:
-                return "No such number"
-
-        return {
-            "email": {
-                **EMAIL_STATUS_FORMATTED,
-                "permanent-failure": _getStatusByBounceSubtype(),
-            },
-            "sms": {
-                **SMS_STATUS_FORMATTED,
-                "provider-failure": _get_sms_status_by_feedback_reason(),
-            },
-            "letter": {
-                "technical-failure": "Technical failure",
-                "sending": "Accepted",
-                "created": "Accepted",
-                "delivered": "Received",
-                "returned-letter": "Returned",
-            },
-        }[self.template.template_type].get(self.status, self.status)
+        return compute_formatted_status(
+            self.template.template_type,
+            self.status,
+            self.feedback_subtype,
+            self.feedback_reason,
+        )
 
     def get_letter_status(self):
         """

app/types.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from datetime import datetime
 from typing import Optional
 
@@ -15,3 +16,18 @@ class VerifiedNotification(NotificationDictToSign):
     created_at: datetime
     job_id: Optional[Job]
     job_row_number: Optional[int]
+
+
+@dataclass
+class NotificationCallbackData:
+    id: str
+    to: str
+    status: str
+    formatted_status: str
+    notification_type: str
+    client_reference: Optional[str]
+    provider_response: Optional[str]
+    created_at: datetime
+    updated_at: Optional[datetime]
+    sent_at: Optional[datetime]
+    service_id: str  # needed by the callback dispatch in nightly_tasks

tests/app/celery/test_nightly_tasks.py

@@ -244,7 +244,7 @@ def test_timeout_notifications_sends_status_update_to_service(client, sample_tem
     timeout_notifications()
 
     signed_data = create_delivery_status_callback_data(notification, callback_api)
-    mocked.assert_called_once_with([str(notification.id), signed_data, notification.service_id], queue=QueueNames.CALLBACKS)
+    mocked.assert_called_once_with([str(notification.id), signed_data, str(notification.service_id)], queue=QueueNames.CALLBACKS)
 
 
 def test_timeout_notifications_logs_and_increments_statsd_for_temporary_failures(notify_api, sample_template, mocker):

tests/app/dao/notification_dao/test_notification_dao.py

@@ -757,10 +757,11 @@ def test_dao_timeout_notifications(sample_template):
         pending = save_notification(create_notification(sample_template, status="pending"))
         delivered = save_notification(create_notification(sample_template, status="delivered"))
 
-    assert Notification.query.get(created.id).status == "created"
-    assert Notification.query.get(sending.id).status == "sending"
-    assert Notification.query.get(pending.id).status == "pending"
-    assert Notification.query.get(delivered.id).status == "delivered"
+        assert Notification.query.get(created.id).status == "created"
+        assert Notification.query.get(sending.id).status == "sending"
+        assert Notification.query.get(pending.id).status == "pending"
+        assert Notification.query.get(delivered.id).status == "delivered"
+
     (
         technical_failure_notifications,
         temporary_failure_notifications,
@@ -771,6 +772,38 @@ def test_dao_timeout_notifications(sample_template):
     assert Notification.query.get(delivered.id).status == "delivered"
     assert len(technical_failure_notifications + temporary_failure_notifications) == 3
 
+    # Check formatted_status for returned callback data
+    for cb in technical_failure_notifications:
+        assert cb.status == "technical-failure"
+        assert cb.formatted_status == "Tech issue"  # for email/sms, technical-failure -> Tech issue
+    for cb in temporary_failure_notifications:
+        assert cb.status == "temporary-failure"
+        assert cb.formatted_status in ("Carrier issue", "Content or inbox issue")  # sms/email
+
+
+def test_timeout_notifications_formatted_status_variants(sample_template):
+    """Test that compute_formatted_status returns correct values for edge cases."""
+    from app.models import compute_formatted_status
+
+    # Permanent failure with feedback_subtype
+    assert compute_formatted_status("email", "permanent-failure", "suppressed", None) == "Blocked"
+    assert compute_formatted_status("email", "permanent-failure", "on-account-suppression-list", None) == "Blocked"
+    # SMS provider-failure with feedback_reason
+    assert (
+        compute_formatted_status("sms", "provider-failure", None, "DESTINATION_COUNTRY_BLOCKED")
+        == "Can't send to this international number"
+    )
+    assert (
+        compute_formatted_status("sms", "provider-failure", None, "NO_ORIGINATION_IDENTITIES_FOUND")
+        == "Can't send to this international number"
+    )
+    # Fallbacks
+    assert compute_formatted_status("sms", "provider-failure", None, None) == "No such number"
+    assert compute_formatted_status("email", "permanent-failure", None, None) == "No such address"
+    # Letter
+    assert compute_formatted_status("letter", "created") == "Accepted"
+    assert compute_formatted_status("letter", "delivered") == "Received"
+
 
 def test_dao_timeout_notifications_only_updates_for_older_notifications(
     sample_template,