feat: Get most recent usage date for inbound number associated to ser…

…vice

- Look through the following tables to get most recent date for inbound sms use: notification, inbound sms and inbound sms history
- Endpoint created to showcase information on the front-end in regards to number activity to prevent us from going into the DB to check the latest use for the number when deciding to archive or release this number

app/dao/inbound_sms_dao.py

@@ -1,3 +1,6 @@
+from datetime import datetime
+from uuid import UUID
+
 from flask import current_app
 from sqlalchemy import and_, desc
 from sqlalchemy.dialects.postgresql import insert
@@ -7,8 +10,10 @@
 from app.constants import SMS_TYPE
 from app.dao.dao_utils import autocommit
 from app.models import (
+    InboundNumber,
     InboundSms,
     InboundSmsHistory,
+    Notification,
     Service,
     ServiceDataRetention,
 )
@@ -179,3 +184,46 @@ def dao_get_paginated_most_recent_inbound_sms_by_user_number_for_service(service
     )
 
     return q.paginate(page=page, per_page=current_app.config["PAGE_SIZE"])
+
+
+def dao_get_most_recent_inbound_usage_date(service_id: UUID, inbound: InboundNumber) -> datetime | None:
+    last_notification = (
+        Notification.query.filter(
+            Notification.reply_to_text == inbound.number,
+            Notification.service_id == service_id,
+        )
+        .order_by(Notification.created_at.desc())
+        .first()
+    )
+
+    last_inbound_sms = (
+        InboundSms.query.filter(
+            InboundSms.notify_number == inbound.number,
+            InboundSms.service_id == service_id,
+        )
+        .order_by(InboundSms.created_at.desc())
+        .first()
+    )
+
+    last_inbound_sms_history = (
+        InboundSmsHistory.query.filter(
+            InboundSmsHistory.notify_number == inbound.number,
+            InboundSmsHistory.service_id == service_id,
+        )
+        .order_by(InboundSmsHistory.created_at.desc())
+        .first()
+    )
+
+    most_recent_usage = max(
+        filter(
+            None,
+            [
+                last_notification.created_at if last_notification else None,
+                last_inbound_sms.created_at if last_inbound_sms else None,
+                last_inbound_sms_history.created_at if last_inbound_sms_history else None,
+            ],
+        ),
+        default=None,
+    )
+
+    return most_recent_usage

app/inbound_sms/rest.py

@@ -2,12 +2,14 @@
 from notifications_utils.recipient_validation.phone_number import try_validate_and_format_phone_number
 
 from app.dao.inbound_numbers_dao import (
+    dao_get_inbound_number_for_service,
     dao_remove_inbound_sms_for_service,
 )
 from app.dao.inbound_sms_dao import (
     dao_count_inbound_sms_for_service,
     dao_get_inbound_sms_by_id,
     dao_get_inbound_sms_for_service,
+    dao_get_most_recent_inbound_usage_date,
     dao_get_paginated_most_recent_inbound_sms_by_user_number_for_service,
 )
 from app.dao.service_data_retention_dao import fetch_service_data_retention_by_notification_type
@@ -82,3 +84,15 @@ def remove_inbound_sms_for_service(service_id):
     except Exception as e:
         current_app.logger.error("error removing inbound SMS for service %s: %s", service_id, e)
         return jsonify({"message": str(e)}), 500
+
+
+@inbound_sms.route("/most-recent-usage", methods=["GET"])
+def get_most_recent_inbound_usage_date(service_id):
+    inbound = dao_get_inbound_number_for_service(service_id)
+
+    if not inbound:
+        return jsonify(message="inbound not found"), 404
+
+    most_recent_date = dao_get_most_recent_inbound_usage_date(service_id, inbound)
+
+    return jsonify(most_recent_date=most_recent_date), 200

tests/app/dao/test_inbound_sms_dao.py

@@ -4,19 +4,25 @@
 from freezegun import freeze_time
 
 from app import db
+from app.constants import KEY_TYPE_NORMAL
+from app.dao.inbound_numbers_dao import dao_get_inbound_number_for_service
 from app.dao.inbound_sms_dao import (
     dao_count_inbound_sms_for_service,
     dao_get_inbound_sms_by_id,
     dao_get_inbound_sms_for_service,
+    dao_get_most_recent_inbound_usage_date,
     dao_get_paginated_inbound_sms_for_service_for_public_api,
     dao_get_paginated_most_recent_inbound_sms_by_user_number_for_service,
     delete_inbound_sms_older_than_retention,
 )
 from app.models import InboundSmsHistory
 from tests.app.db import (
     create_inbound_sms,
+    create_job,
+    create_notification,
     create_service,
     create_service_data_retention,
+    create_template,
 )
 from tests.conftest import set_config
 
@@ -353,3 +359,54 @@ def test_most_recent_inbound_sms_only_returns_values_within_7_days(sample_servic
 
     assert len(res.items) == 1
     assert res.items[0].content == "new"
+
+
+def test_dao_get_most_recent_inbound_usage_date_notifications_table(sample_service, sample_inbound_numbers):
+    template = create_template(service=sample_service)
+    job = create_job(template=template)
+    inbound = dao_get_inbound_number_for_service(sample_service.id)
+
+    notification_time = datetime.utcnow()
+
+    create_notification(
+        template=template,
+        job=job,
+        job_row_number=None,
+        to_field=None,
+        status="created",
+        reference=None,
+        created_at=notification_time,
+        sent_at=None,
+        billable_units=1,
+        personalisation=None,
+        api_key=None,
+        key_type=KEY_TYPE_NORMAL,
+        reply_to_text=inbound.number,
+    )
+
+    most_recent_date = dao_get_most_recent_inbound_usage_date(sample_service.id, inbound)
+    assert most_recent_date == notification_time
+
+
+def test_dao_get_most_recent_inbound_usage_date_inbound_sms_table(sample_service, sample_inbound_numbers):
+    sms_time = datetime.utcnow()
+    create_inbound_sms(sample_service, created_at=sms_time)
+    inbound = dao_get_inbound_number_for_service(sample_service.id)
+
+    most_recent_date = dao_get_most_recent_inbound_usage_date(sample_service.id, inbound)
+    assert most_recent_date == sms_time
+
+
+def test_dao_get_most_recent_inbound_usage_date_inbound_sms_history_table(
+    sample_service, sample_inbound_numbers, sample_inbound_sms_history
+):
+    inbound = next((x for x in sample_inbound_numbers if x.service_id == sample_service.id), None)
+    most_recent_date = dao_get_most_recent_inbound_usage_date(sample_service.id, inbound)
+
+    assert most_recent_date is not None
+    assert most_recent_date.date() == sample_inbound_sms_history.created_at.date()
+
+
+def test_dao_get_most_recent_inbound_usage_date_no_recent_notifications(sample_service, sample_inbound_numbers):
+    inbound = next((x for x in sample_inbound_numbers if x.service_id == sample_service.id), None)
+    assert dao_get_most_recent_inbound_usage_date(sample_service.id, inbound) is None

tests/app/inbound_sms/test_rest.py

@@ -2,6 +2,7 @@
 from datetime import datetime, timedelta
 
 import pytest
+from dateutil.parser import parse
 from freezegun import freeze_time
 
 from app.constants import INBOUND_SMS_TYPE
@@ -314,3 +315,38 @@ def test_remove_inbound_sms_for_service_success_without_inbound_number(admin_req
         _data={"archive": True},
         _expected_status=200,
     )
+
+
+def test_get_most_recent_inbound_usage_date_success(
+    admin_request, sample_service, sample_inbound_numbers, sample_inbound_sms_history
+):
+    response = admin_request.get(
+        "inbound_sms.get_most_recent_inbound_usage_date", service_id=sample_service.id, _expected_status=200
+    )
+
+    assert response is not None
+    response_date = parse(response["most_recent_date"])
+
+    assert response_date.date() == datetime.utcnow().date()
+
+
+def test_get_most_recent_inbound_usage_date_success_no_usage_found(
+    admin_request, sample_service, sample_inbound_numbers
+):
+    response = admin_request.get(
+        "inbound_sms.get_most_recent_inbound_usage_date", service_id=sample_service.id, _expected_status=200
+    )
+
+    assert response is not None
+    assert response["most_recent_date"] is None
+
+
+def test_get_most_recent_inbound_usage_date_404_no_inbound(
+    admin_request,
+    sample_service,
+):
+    response = admin_request.get(
+        "inbound_sms.get_most_recent_inbound_usage_date", service_id=sample_service.id, _expected_status=404
+    )
+
+    assert response["message"] == "inbound not found"