feat: message group id for callback entry points (SES, inbound SMS)

Author: joybytes

app/notifications/notifications_ses_callback.py

@@ -74,7 +74,9 @@ def check_and_queue_callback_task(notification):
     if service_callback_api:
         notification_data = create_delivery_status_callback_data(notification, service_callback_api)
         send_delivery_status_to_service.apply_async(
-            [str(notification.id), notification_data], queue=QueueNames.CALLBACKS
+            [str(notification.id), notification_data],
+            queue=QueueNames.CALLBACKS,
+            MessageGroupId=str(notification.service_id),
         )
 
 
@@ -83,4 +85,8 @@ def _check_and_queue_complaint_callback_task(complaint, notification, recipient)
     service_callback_api = get_complaint_callback_api_for_service(service_id=notification.service_id)
     if service_callback_api:
         complaint_data = create_complaint_callback_data(complaint, notification, service_callback_api, recipient)
-        send_complaint_to_service.apply_async([complaint_data], queue=QueueNames.CALLBACKS)
+        send_complaint_to_service.apply_async(
+            [complaint_data],
+            queue=QueueNames.CALLBACKS,
+            MessageGroupId=str(notification.service_id),
+        )

app/notifications/receive_notifications.py

@@ -69,7 +69,9 @@ def receive_mmg_sms():
     )
 
     service_callback_tasks.send_inbound_sms_to_service.apply_async(
-        [str(inbound.id), str(service.id)], queue=QueueNames.CALLBACKS
+        [str(inbound.id), str(service.id)],
+        queue=QueueNames.CALLBACKS,
+        MessageGroupId=str(service.id),
     )
 
     current_app.logger.info(
@@ -115,7 +117,9 @@ def receive_firetext_sms():
     INBOUND_SMS_COUNTER.labels("firetext").inc()
 
     service_callback_tasks.send_inbound_sms_to_service.apply_async(
-        [str(inbound.id), str(service.id)], queue=QueueNames.CALLBACKS
+        [str(inbound.id), str(service.id)],
+        queue=QueueNames.CALLBACKS,
+        MessageGroupId=str(service.id),
     )
     current_app.logger.info(
         "%s received inbound SMS with reference %s from Firetext",

tests/app/notifications/test_notifications_ses_callback.py

@@ -91,7 +91,9 @@ def test_check_and_queue_callback_task(mocker, mock_celery_task, sample_notifica
     assert mock_create_args[1].id == callback_api.id
 
     mock_send.assert_called_once_with(
-        [str(sample_notification.id), mock_create.return_value], queue="service-callbacks"
+        [str(sample_notification.id), mock_create.return_value],
+        queue="service-callbacks",
+        MessageGroupId=str(sample_notification.service_id),
     )
 
 

tests/app/notifications/test_receive_notification.py

@@ -72,7 +72,9 @@ def test_receive_notification_returns_received_to_mmg(client, mocker, sample_ser
 
     inbound_sms_id = InboundSms.query.all()[0].id
     mocked.assert_called_once_with(
-        [str(inbound_sms_id), str(sample_service_full_permissions.id)], queue="service-callbacks"
+        [str(inbound_sms_id), str(sample_service_full_permissions.id)],
+        queue="service-callbacks",
+        MessageGroupId=str(sample_service_full_permissions.id),
     )
 
 
@@ -340,7 +342,11 @@ def test_receive_notification_returns_received_to_firetext(notify_db_session, cl
 
     assert result["status"] == "ok"
     inbound_sms_id = InboundSms.query.all()[0].id
-    mocked.assert_called_once_with([str(inbound_sms_id), str(service.id)], queue="service-callbacks")
+    mocked.assert_called_once_with(
+        [str(inbound_sms_id), str(service.id)],
+        queue="service-callbacks",
+        MessageGroupId=str(service.id),
+    )
 
 
 def test_receive_notification_from_firetext_persists_message(notify_db_session, client, mocker):
@@ -368,7 +374,11 @@ def test_receive_notification_from_firetext_persists_message(notify_db_session,
     assert persisted.content == "this is a message"
     assert persisted.provider == "firetext"
     assert persisted.provider_date == datetime(2017, 1, 1, 12, 0, 0, 0)
-    mocked.assert_called_once_with([str(persisted.id), str(service.id)], queue="service-callbacks")
+    mocked.assert_called_once_with(
+        [str(persisted.id), str(service.id)],
+        queue="service-callbacks",
+        MessageGroupId=str(service.id),
+    )
 
 
 def test_receive_notification_from_firetext_persists_message_with_normalized_phone(notify_db_session, client, mocker):