From cc25bb7db80ba2b45bbaab474cd792e28f1c13a6 Mon Sep 17 00:00:00 2001 From: tenkus47 Date: Mon, 24 Aug 2026 15:14:14 +0530 Subject: [PATCH 1/2] Add group post notification features - Implemented the `fetch_group_post_notification_targets` function in the backend client to retrieve notification targets for group posts. - Added schemas for group post notifications, including `GroupPostNotificationTargetsResponse` and related models. - Created functions for building and sending group post notifications, including `build_group_post_notification_data` and `send_group_post_push_notification`. - Introduced unit tests for the new group post notification functionalities, ensuring proper response parsing and payload structure. - Updated configuration settings for group post notifications in the application. --- tests/notifications/test_backend_client.py | 67 +++++ tests/notifications/test_fcm_client.py | 51 ++++ .../test_group_post_notification_consumer.py | 220 +++++++++++++++ .../test_group_post_sqs_client.py | 114 ++++++++ worker_api/config.py | 11 + worker_api/db/mongo_database.py | 4 + .../notifications/group_post_sqs_client.py | 98 +++++++ worker_api/notifications/schemas.py | 24 ++ .../notifications/services/backend_client.py | 17 ++ .../group_post_notification_consumer.py | 257 ++++++++++++++++++ .../notifications/services/push/fcm_client.py | 45 +++ 11 files changed, 908 insertions(+) create mode 100644 tests/notifications/test_group_post_notification_consumer.py create mode 100644 tests/notifications/test_group_post_sqs_client.py create mode 100644 worker_api/notifications/group_post_sqs_client.py create mode 100644 worker_api/notifications/services/group_post_notification_consumer.py diff --git a/tests/notifications/test_backend_client.py b/tests/notifications/test_backend_client.py index d1d9c10..7be4147 100644 --- a/tests/notifications/test_backend_client.py +++ b/tests/notifications/test_backend_client.py @@ -197,6 +197,73 @@ async def test_defaults_to_first_page(self): assert http_client.get.await_args.kwargs["params"] == {"skip": 0, "limit": 100} +class TestFetchGroupPostNotificationTargets: + @pytest.mark.asyncio + async def test_returns_parsed_targets(self): + post_id = uuid4() + device_id = uuid4() + response = _json_response( + { + "post_id": str(post_id), + "group_id": str(uuid4()), + "author_id": str(uuid4()), + "title": "Sangha", + "body": "Alice shared a new post", + "recipients": [ + { + "user_id": str(uuid4()), + "push_devices": [ + {"id": str(device_id), "token": "token-1", "platform": "ios"} + ], + } + ], + "skip": 100, + "limit": 50, + "total": 120, + "has_more": False, + } + ) + client_patch, http_client = _patch_async_client(response) + + with client_patch, _patch_config(): + targets = await backend_client.fetch_group_post_notification_targets( + post_id=post_id, + skip=100, + limit=50, + ) + + assert targets.total == 120 + assert targets.recipients[0].push_devices[0].id == device_id + assert http_client.get.await_args.args[0] == ( + f"http://backend.test/internal/group-post-notification-targets/{post_id}" + ) + assert http_client.get.await_args.kwargs["params"] == {"skip": 100, "limit": 50} + + @pytest.mark.asyncio + async def test_defaults_to_first_page(self): + post_id = uuid4() + response = _json_response( + { + "post_id": str(post_id), + "group_id": str(uuid4()), + "author_id": str(uuid4()), + "title": "Sangha", + "body": "Alice shared a new post", + "recipients": [], + "skip": 0, + "limit": 100, + "total": 0, + "has_more": False, + } + ) + client_patch, http_client = _patch_async_client(response) + + with client_patch, _patch_config(): + await backend_client.fetch_group_post_notification_targets(post_id=post_id) + + assert http_client.get.await_args.kwargs["params"] == {"skip": 0, "limit": 100} + + class TestFetchVerseOfDayNotificationTargets: @pytest.mark.asyncio async def test_returns_parsed_targets(self): diff --git a/tests/notifications/test_fcm_client.py b/tests/notifications/test_fcm_client.py index f690ce6..bcc1d91 100644 --- a/tests/notifications/test_fcm_client.py +++ b/tests/notifications/test_fcm_client.py @@ -5,9 +5,11 @@ from worker_api.notifications.services.push.fcm_client import ( build_chat_notification_data, + build_group_post_notification_data, build_routine_notification_data, send_chat_push_notification, send_fcm_notification, + send_group_post_push_notification, send_routine_push_notification, ) @@ -89,6 +91,31 @@ def test_empty_group_id_for_private(self): assert data["group_id"] == "" +class TestBuildGroupPostNotificationData: + def test_includes_group_post_routing_fields(self): + post_id = uuid4() + group_id = uuid4() + author_id = uuid4() + data = build_group_post_notification_data( + post_id=post_id, + group_id=group_id, + author_id=author_id, + title="Sangha", + body="Alice shared a new post", + ) + assert data == { + "notification_type": "GROUP_POST", + "session_type": "GROUP_POST", + "post_id": str(post_id), + "group_id": str(group_id), + "author_id": str(author_id), + "source_id": str(post_id), + "title": "Sangha", + "body": "Alice shared a new post", + "image_url": "", + } + + class TestSendFcmNotification: @pytest.mark.asyncio @patch("worker_api.notifications.services.push.fcm_client.messaging.send") @@ -184,3 +211,27 @@ async def test_delegates_with_chat_payload(self, mock_send): assert kwargs["data"]["notification_type"] == "CHAT_MESSAGE" assert kwargs["data"]["session_type"] == "CHAT" assert kwargs["data"]["room_id"] == str(room_id) + + +class TestSendGroupPostPushNotification: + @pytest.mark.asyncio + @patch("worker_api.notifications.services.push.fcm_client.send_fcm_notification") + async def test_delegates_with_group_post_payload(self, mock_send): + post_id = uuid4() + group_id = uuid4() + author_id = uuid4() + await send_group_post_push_notification( + device_token="device-token", + post_id=post_id, + group_id=group_id, + author_id=author_id, + title="Sangha", + body="Alice shared a new post", + ) + mock_send.assert_awaited_once() + kwargs = mock_send.await_args.kwargs + assert kwargs["title"] == "Sangha" + assert kwargs["body"] == "Alice shared a new post" + assert kwargs["data"]["notification_type"] == "GROUP_POST" + assert kwargs["data"]["session_type"] == "GROUP_POST" + assert kwargs["data"]["post_id"] == str(post_id) diff --git a/tests/notifications/test_group_post_notification_consumer.py b/tests/notifications/test_group_post_notification_consumer.py new file mode 100644 index 0000000..f6819d4 --- /dev/null +++ b/tests/notifications/test_group_post_notification_consumer.py @@ -0,0 +1,220 @@ +"""Tests for group post notification SQS consumer.""" +import json +from unittest.mock import AsyncMock, patch +from uuid import uuid4 + +import pytest + +from worker_api.notifications.schemas import ( + GroupPostNotificationRecipient, + GroupPostNotificationTargetsResponse, + GroupPostPushDeviceTarget, +) +from worker_api.notifications.services.group_post_notification_consumer import ( + TransientGroupPostNotificationError, + process_group_post_notification_message, +) +from worker_api.notifications.services.push.fcm_client import PermanentPushTokenError + + +def _targets(*, post_id, devices): + return GroupPostNotificationTargetsResponse( + post_id=post_id, + group_id=uuid4(), + author_id=uuid4(), + title="Sangha", + body="Alice shared a new post", + recipients=[ + GroupPostNotificationRecipient( + user_id=uuid4(), + push_devices=devices, + ) + ], + skip=0, + limit=100, + total=1, + has_more=False, + ) + + +def _body(post_id): + return json.dumps( + { + "event_type": "GROUP_POST_CREATED", + "version": 1, + "post_id": str(post_id), + } + ) + + +class TestProcessGroupPostNotificationMessage: + @pytest.mark.asyncio + @patch("worker_api.notifications.services.group_post_notification_consumer.delete_group_post_notification_message") + async def test_deletes_malformed_message(self, mock_delete): + await process_group_post_notification_message( + {"ReceiptHandle": "r1", "Body": "not-json"} + ) + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch("worker_api.notifications.services.group_post_notification_consumer.delete_group_post_notification_message") + @patch( + "worker_api.notifications.services.group_post_notification_consumer._fetch_all_targets", + new_callable=AsyncMock, + ) + @patch("worker_api.notifications.services.group_post_notification_consumer.get_bool", return_value=True) + async def test_deletes_when_post_not_found(self, _get_bool, mock_fetch, mock_delete): + from fastapi import HTTPException + + mock_fetch.side_effect = HTTPException(status_code=404, detail="not found") + post_id = uuid4() + await process_group_post_notification_message( + {"ReceiptHandle": "r1", "Body": _body(post_id)} + ) + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch("worker_api.notifications.services.group_post_notification_consumer.delete_group_post_notification_message") + @patch( + "worker_api.notifications.services.group_post_notification_consumer.send_group_post_push_notification", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer._fetch_all_targets", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer.is_push_configured", + return_value=True, + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer._already_sent", + return_value=False, + ) + @patch("worker_api.notifications.services.group_post_notification_consumer._mark_sent") + @patch("worker_api.notifications.services.group_post_notification_consumer.get_int", return_value=5) + @patch("worker_api.notifications.services.group_post_notification_consumer.get_bool", return_value=True) + async def test_sends_and_deletes_on_success( + self, _get_bool, _get_int, mock_mark, _already, _configured, mock_fetch, mock_send, mock_delete, + ): + post_id = uuid4() + device = GroupPostPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(post_id=post_id, devices=[device]) + + await process_group_post_notification_message( + {"ReceiptHandle": "r1", "Body": _body(post_id)} + ) + + mock_send.assert_awaited_once() + mock_mark.assert_called_once() + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch("worker_api.notifications.services.group_post_notification_consumer.delete_group_post_notification_message") + @patch( + "worker_api.notifications.services.group_post_notification_consumer.deactivate_push_device", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer.send_group_post_push_notification", + new_callable=AsyncMock, + side_effect=PermanentPushTokenError("gone"), + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer._fetch_all_targets", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer.is_push_configured", + return_value=True, + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer._already_sent", + return_value=False, + ) + @patch("worker_api.notifications.services.group_post_notification_consumer._mark_sent") + @patch("worker_api.notifications.services.group_post_notification_consumer.get_int", return_value=5) + @patch("worker_api.notifications.services.group_post_notification_consumer.get_bool", return_value=True) + async def test_permanent_token_deactivates_and_deletes( + self, _get_bool, _get_int, mock_mark, _already, _configured, mock_fetch, mock_send, mock_deactivate, mock_delete, + ): + post_id = uuid4() + device = GroupPostPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(post_id=post_id, devices=[device]) + + await process_group_post_notification_message( + {"ReceiptHandle": "r1", "Body": _body(post_id)} + ) + + mock_deactivate.assert_awaited_once_with(push_device_id=device.id) + mock_mark.assert_called_once() + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch("worker_api.notifications.services.group_post_notification_consumer.delete_group_post_notification_message") + @patch( + "worker_api.notifications.services.group_post_notification_consumer.send_group_post_push_notification", + new_callable=AsyncMock, + side_effect=RuntimeError("temporary"), + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer._fetch_all_targets", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer.is_push_configured", + return_value=True, + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer._already_sent", + return_value=False, + ) + @patch("worker_api.notifications.services.group_post_notification_consumer.get_int", return_value=5) + @patch("worker_api.notifications.services.group_post_notification_consumer.get_bool", return_value=True) + async def test_transient_failure_leaves_message( + self, _get_bool, _get_int, _already, _configured, mock_fetch, mock_send, mock_delete, + ): + post_id = uuid4() + device = GroupPostPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(post_id=post_id, devices=[device]) + + with pytest.raises(TransientGroupPostNotificationError): + await process_group_post_notification_message( + {"ReceiptHandle": "r1", "Body": _body(post_id)} + ) + + mock_delete.assert_not_called() + + @pytest.mark.asyncio + @patch("worker_api.notifications.services.group_post_notification_consumer.delete_group_post_notification_message") + @patch( + "worker_api.notifications.services.group_post_notification_consumer.send_group_post_push_notification", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer._fetch_all_targets", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer.is_push_configured", + return_value=True, + ) + @patch( + "worker_api.notifications.services.group_post_notification_consumer._already_sent", + return_value=True, + ) + @patch("worker_api.notifications.services.group_post_notification_consumer.get_int", return_value=5) + @patch("worker_api.notifications.services.group_post_notification_consumer.get_bool", return_value=True) + async def test_skips_already_sent_devices( + self, _get_bool, _get_int, _already, _configured, mock_fetch, mock_send, mock_delete, + ): + post_id = uuid4() + device = GroupPostPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(post_id=post_id, devices=[device]) + + await process_group_post_notification_message( + {"ReceiptHandle": "r1", "Body": _body(post_id)} + ) + + mock_send.assert_not_called() + mock_delete.assert_called_once_with("r1") diff --git a/tests/notifications/test_group_post_sqs_client.py b/tests/notifications/test_group_post_sqs_client.py new file mode 100644 index 0000000..1ad22c1 --- /dev/null +++ b/tests/notifications/test_group_post_sqs_client.py @@ -0,0 +1,114 @@ +"""Tests for group post notification SQS client helpers.""" +import json +from unittest.mock import MagicMock, patch +from uuid import uuid4 + +from botocore.exceptions import ClientError + +from worker_api.notifications.group_post_sqs_client import ( + GROUP_POST_CREATED_EVENT, + GROUP_POST_NOTIFICATION_EVENT_VERSION, + delete_group_post_notification_message, + get_group_post_notification_sqs_queue_url, + is_group_post_notification_sqs_configured, + is_group_post_notification_sqs_poll_enabled, + parse_group_post_notification_message_body, + receive_group_post_notification_messages, +) + + +class TestQueueConfig: + @patch("worker_api.notifications.group_post_sqs_client.get", return_value=" https://sqs.example/group-post ") + def test_get_queue_url_strips(self, _get): + assert get_group_post_notification_sqs_queue_url() == "https://sqs.example/group-post" + + @patch("worker_api.notifications.group_post_sqs_client.get_group_post_notification_sqs_queue_url", return_value="") + def test_is_configured_false(self, _url): + assert is_group_post_notification_sqs_configured() is False + + @patch("worker_api.notifications.group_post_sqs_client.is_group_post_notification_sqs_configured", return_value=True) + @patch("worker_api.notifications.group_post_sqs_client.get_bool", return_value=True) + def test_poll_enabled(self, _bool, _configured): + assert is_group_post_notification_sqs_poll_enabled() is True + + +class TestParseBody: + def test_valid_event(self): + post_id = str(uuid4()) + body = parse_group_post_notification_message_body( + json.dumps( + { + "event_type": GROUP_POST_CREATED_EVENT, + "version": GROUP_POST_NOTIFICATION_EVENT_VERSION, + "post_id": post_id, + } + ) + ) + assert body["post_id"] == post_id + + def test_rejects_invalid_json(self): + assert parse_group_post_notification_message_body("not-json") is None + + def test_rejects_wrong_event_type(self): + assert parse_group_post_notification_message_body( + json.dumps({"event_type": "OTHER", "version": 1, "post_id": str(uuid4())}) + ) is None + + def test_rejects_wrong_version(self): + assert parse_group_post_notification_message_body( + json.dumps( + { + "event_type": GROUP_POST_CREATED_EVENT, + "version": 99, + "post_id": str(uuid4()), + } + ) + ) is None + + def test_rejects_missing_post_id(self): + assert parse_group_post_notification_message_body( + json.dumps({"event_type": GROUP_POST_CREATED_EVENT, "version": 1}) + ) is None + + +class TestReceiveAndDelete: + @patch("worker_api.notifications.group_post_sqs_client.get_group_post_notification_sqs_queue_url", return_value="") + def test_receive_empty_without_queue(self, _url): + assert receive_group_post_notification_messages() == [] + + @patch("worker_api.notifications.group_post_sqs_client.get_int", side_effect=lambda key: 1) + @patch("worker_api.notifications.group_post_sqs_client._get_sqs_client") + @patch( + "worker_api.notifications.group_post_sqs_client.get_group_post_notification_sqs_queue_url", + return_value="https://sqs.example/group-post", + ) + def test_receive_messages(self, _url, mock_get_client, _get_int): + client = MagicMock() + client.receive_message.return_value = {"Messages": [{"MessageId": "1"}]} + mock_get_client.return_value = client + assert receive_group_post_notification_messages() == [{"MessageId": "1"}] + + @patch("worker_api.notifications.group_post_sqs_client._get_sqs_client") + @patch( + "worker_api.notifications.group_post_sqs_client.get_group_post_notification_sqs_queue_url", + return_value="https://sqs.example/group-post", + ) + def test_delete_message(self, _url, mock_get_client): + client = MagicMock() + mock_get_client.return_value = client + delete_group_post_notification_message("receipt") + client.delete_message.assert_called_once() + + @patch("worker_api.notifications.group_post_sqs_client._get_sqs_client") + @patch( + "worker_api.notifications.group_post_sqs_client.get_group_post_notification_sqs_queue_url", + return_value="https://sqs.example/group-post", + ) + def test_delete_handles_client_error(self, _url, mock_get_client): + client = MagicMock() + client.delete_message.side_effect = ClientError( + {"Error": {"Code": "500", "Message": "fail"}}, + "DeleteMessage", + ) + mock_get_client.return_value = client + delete_group_post_notification_message("receipt") diff --git a/worker_api/config.py b/worker_api/config.py index 74c6456..f1d8f9a 100644 --- a/worker_api/config.py +++ b/worker_api/config.py @@ -58,6 +58,17 @@ JOIN_REQUEST_NOTIFICATION_IDEMPOTENCY_TTL_SECONDS=86400, JOIN_REQUEST_NOTIFICATION_IDEMPOTENCY_KEY_PREFIX="worker:join-request-notifications:sent:", + # Group post notification SQS consumer (backend producer → worker consumer) + GROUP_POST_NOTIFICATION_SQS_QUEUE_URL="", + GROUP_POST_NOTIFICATION_SQS_WAIT_TIME_SECONDS=20, + GROUP_POST_NOTIFICATION_SQS_VISIBILITY_TIMEOUT_SECONDS=300, + GROUP_POST_NOTIFICATION_SQS_MAX_MESSAGES=5, + GROUP_POST_NOTIFICATION_SQS_POLL_ENABLED="true", + GROUP_POST_NOTIFICATION_SEND_CONCURRENCY=10, + GROUP_POST_NOTIFICATION_TARGET_PAGE_SIZE=100, + GROUP_POST_NOTIFICATION_IDEMPOTENCY_TTL_SECONDS=86400, + GROUP_POST_NOTIFICATION_IDEMPOTENCY_KEY_PREFIX="worker:group-post-notifications:sent:", + # Notification dispatch (Cloud Scheduler -> worker) NOTIFICATION_DISPATCH_SECRET_TOKEN="Dispatch", NOTIFICATION_DISPATCH_BATCH_SIZE=100, diff --git a/worker_api/db/mongo_database.py b/worker_api/db/mongo_database.py index 6710487..7fc55a8 100644 --- a/worker_api/db/mongo_database.py +++ b/worker_api/db/mongo_database.py @@ -11,6 +11,9 @@ from worker_api.notifications.services.chat_notification_consumer import ( run_chat_notification_sqs_consumer, ) +from worker_api.notifications.services.group_post_notification_consumer import ( + run_group_post_notification_sqs_consumer, +) from worker_api.notifications.services.join_request_notification_consumer import ( run_join_request_notification_sqs_consumer, ) @@ -42,6 +45,7 @@ async def lifespan(api: FastAPI): asyncio.create_task(run_audio_sqs_consumer(stop_event)), asyncio.create_task(run_chat_notification_sqs_consumer(stop_event)), asyncio.create_task(run_join_request_notification_sqs_consumer(stop_event)), + asyncio.create_task(run_group_post_notification_sqs_consumer(stop_event)), ] yield diff --git a/worker_api/notifications/group_post_sqs_client.py b/worker_api/notifications/group_post_sqs_client.py new file mode 100644 index 0000000..1d48a2a --- /dev/null +++ b/worker_api/notifications/group_post_sqs_client.py @@ -0,0 +1,98 @@ +import json +import logging +from typing import Any, Dict, List, Optional + +import boto3 +from botocore.exceptions import ClientError + +from worker_api.config import get, get_bool, get_int + +logger = logging.getLogger(__name__) + +_sqs_client = None + +GROUP_POST_CREATED_EVENT = "GROUP_POST_CREATED" +GROUP_POST_NOTIFICATION_EVENT_VERSION = 1 + + +def _get_sqs_client(): + global _sqs_client + if _sqs_client is None: + _sqs_client = boto3.client( + "sqs", + aws_access_key_id=get("AWS_ACCESS_KEY"), + aws_secret_access_key=get("AWS_SECRET_KEY"), + region_name=get("AWS_REGION"), + ) + return _sqs_client + + +def get_group_post_notification_sqs_queue_url() -> str: + return get("GROUP_POST_NOTIFICATION_SQS_QUEUE_URL").strip() + + +def is_group_post_notification_sqs_configured() -> bool: + return bool(get_group_post_notification_sqs_queue_url()) + + +def is_group_post_notification_sqs_poll_enabled() -> bool: + return get_bool("GROUP_POST_NOTIFICATION_SQS_POLL_ENABLED") and is_group_post_notification_sqs_configured() + + +def receive_group_post_notification_messages() -> List[Dict[str, Any]]: + queue_url = get_group_post_notification_sqs_queue_url() + if not queue_url: + return [] + + try: + response = _get_sqs_client().receive_message( + QueueUrl=queue_url, + MaxNumberOfMessages=get_int("GROUP_POST_NOTIFICATION_SQS_MAX_MESSAGES"), + WaitTimeSeconds=get_int("GROUP_POST_NOTIFICATION_SQS_WAIT_TIME_SECONDS"), + VisibilityTimeout=get_int("GROUP_POST_NOTIFICATION_SQS_VISIBILITY_TIMEOUT_SECONDS"), + MessageAttributeNames=["All"], + ) + return response.get("Messages", []) + except ClientError as e: + logger.error("Failed to receive group post notification SQS messages: %s", e) + return [] + + +def delete_group_post_notification_message(receipt_handle: str) -> None: + queue_url = get_group_post_notification_sqs_queue_url() + if not queue_url or not receipt_handle: + return + + try: + _get_sqs_client().delete_message( + QueueUrl=queue_url, + ReceiptHandle=receipt_handle, + ) + except ClientError as e: + logger.error("Failed to delete group post notification SQS message: %s", e) + + +def parse_group_post_notification_message_body(raw_body: str) -> Optional[Dict[str, Any]]: + try: + body = json.loads(raw_body) + except (TypeError, json.JSONDecodeError) as e: + logger.error("Invalid group post notification SQS message body: %s", e) + return None + + if not isinstance(body, dict): + logger.error("Group post notification SQS message body is not an object: %s", body) + return None + + event_type = body.get("event_type") + version = body.get("version") + post_id = body.get("post_id") + if event_type != GROUP_POST_CREATED_EVENT: + logger.error("Unsupported group post notification event_type: %s", event_type) + return None + if version != GROUP_POST_NOTIFICATION_EVENT_VERSION: + logger.error("Unsupported group post notification event version: %s", version) + return None + if not post_id: + logger.error("Group post notification SQS message missing post_id: %s", body) + return None + return body diff --git a/worker_api/notifications/schemas.py b/worker_api/notifications/schemas.py index 9c60d41..348611d 100644 --- a/worker_api/notifications/schemas.py +++ b/worker_api/notifications/schemas.py @@ -193,6 +193,30 @@ class ChatNotificationTargetsResponse(BaseModel): has_more: bool +class GroupPostPushDeviceTarget(BaseModel): + id: UUID + token: str + platform: str + + +class GroupPostNotificationRecipient(BaseModel): + user_id: UUID + push_devices: list[GroupPostPushDeviceTarget] + + +class GroupPostNotificationTargetsResponse(BaseModel): + post_id: UUID + group_id: UUID + author_id: UUID + title: str + body: str + recipients: list[GroupPostNotificationRecipient] + skip: int + limit: int + total: int + has_more: bool + + class JoinRequestPushDeviceTarget(BaseModel): id: UUID token: str diff --git a/worker_api/notifications/services/backend_client.py b/worker_api/notifications/services/backend_client.py index 2fbee04..63821f5 100644 --- a/worker_api/notifications/services/backend_client.py +++ b/worker_api/notifications/services/backend_client.py @@ -6,6 +6,7 @@ from worker_api.notifications.schemas import ( ChatNotificationTargetsResponse, DeactivatePushDeviceResponse, + GroupPostNotificationTargetsResponse, JoinRequestNotificationTargetsResponse, NotificationContent, RoutineNotificationTargetsResponse, @@ -94,6 +95,22 @@ async def fetch_verse_of_day_notification_targets() -> VerseOfDayNotificationTar return VerseOfDayNotificationTargetsResponse.model_validate(response.json()) +async def fetch_group_post_notification_targets( + *, + post_id: UUID, + skip: int = 0, + limit: int = 100, +) -> GroupPostNotificationTargetsResponse: + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.get( + f"{_backend_url()}/internal/group-post-notification-targets/{post_id}", + params={"skip": skip, "limit": limit}, + headers=_backend_headers(), + ) + response.raise_for_status() + return GroupPostNotificationTargetsResponse.model_validate(response.json()) + + async def deactivate_push_device(*, push_device_id: UUID) -> DeactivatePushDeviceResponse: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( diff --git a/worker_api/notifications/services/group_post_notification_consumer.py b/worker_api/notifications/services/group_post_notification_consumer.py new file mode 100644 index 0000000..f97344e --- /dev/null +++ b/worker_api/notifications/services/group_post_notification_consumer.py @@ -0,0 +1,257 @@ +import asyncio +import logging +from typing import Any, Dict, Optional +from uuid import UUID + +import httpx +import redis +from fastapi import HTTPException + +from worker_api.config import get, get_bool, get_int +from worker_api.notifications.group_post_sqs_client import ( + delete_group_post_notification_message, + is_group_post_notification_sqs_poll_enabled, + parse_group_post_notification_message_body, + receive_group_post_notification_messages, +) +from worker_api.notifications.schemas import ( + GroupPostNotificationTargetsResponse, + GroupPostPushDeviceTarget, +) +from worker_api.notifications.services.backend_client import ( + deactivate_push_device, + fetch_group_post_notification_targets, +) +from worker_api.notifications.services.push.config_loader import is_push_configured +from worker_api.notifications.services.push.fcm_client import ( + PermanentPushTokenError, + send_group_post_push_notification, +) + +logger = logging.getLogger(__name__) + +_POLL_IDLE_SECONDS = 5 +_POLL_ERROR_SECONDS = 10 +_redis_client: redis.Redis | None = None + + +class TransientGroupPostNotificationError(Exception): + """Raised when processing should leave the SQS message for retry.""" + + +def _get_redis_client() -> redis.Redis: + global _redis_client + if _redis_client is None: + _redis_client = redis.Redis.from_url(get("CACHE_CONNECTION_STRING")) + return _redis_client + + +def _idempotency_key(*, post_id: UUID, push_device_id: UUID) -> str: + prefix = get("GROUP_POST_NOTIFICATION_IDEMPOTENCY_KEY_PREFIX") + return f"{prefix}{post_id}:{push_device_id}" + + +def _already_sent(*, post_id: UUID, push_device_id: UUID) -> bool: + client = _get_redis_client() + return bool(client.exists(_idempotency_key(post_id=post_id, push_device_id=push_device_id))) + + +def _mark_sent(*, post_id: UUID, push_device_id: UUID) -> None: + client = _get_redis_client() + client.setex( + _idempotency_key(post_id=post_id, push_device_id=push_device_id), + get_int("GROUP_POST_NOTIFICATION_IDEMPOTENCY_TTL_SECONDS"), + "1", + ) + + +def _parse_uuid(value: Any) -> Optional[UUID]: + if value is None or value == "": + return None + return UUID(str(value)) + + +async def _fetch_all_targets(post_id: UUID) -> GroupPostNotificationTargetsResponse: + page_size = max(get_int("GROUP_POST_NOTIFICATION_TARGET_PAGE_SIZE"), 1) + skip = 0 + first_page: GroupPostNotificationTargetsResponse | None = None + all_recipients = [] + + while True: + try: + page = await fetch_group_post_notification_targets( + post_id=post_id, + skip=skip, + limit=page_size, + ) + except httpx.HTTPStatusError as exc: + if exc.response.status_code == 404: + raise HTTPException(status_code=404, detail="Group post not found") from exc + raise TransientGroupPostNotificationError(str(exc)) from exc + except httpx.HTTPError as exc: + raise TransientGroupPostNotificationError(str(exc)) from exc + + if first_page is None: + first_page = page + all_recipients.extend(page.recipients) + if not page.has_more: + break + skip += page.limit + + assert first_page is not None + return first_page.model_copy(update={"recipients": all_recipients, "has_more": False}) + + +async def _send_to_device( + *, + targets: GroupPostNotificationTargetsResponse, + device: GroupPostPushDeviceTarget, + semaphore: asyncio.Semaphore, +) -> str: + """Return sent | skipped | permanent_failed | transient_failed.""" + async with semaphore: + if not is_push_configured(device.platform): + return "skipped" + + if _already_sent(post_id=targets.post_id, push_device_id=device.id): + return "skipped" + + try: + await send_group_post_push_notification( + device_token=device.token, + post_id=targets.post_id, + group_id=targets.group_id, + author_id=targets.author_id, + title=targets.title, + body=targets.body, + ) + _mark_sent(post_id=targets.post_id, push_device_id=device.id) + return "sent" + except PermanentPushTokenError: + logger.warning( + "Deactivating permanently invalid push device %s for group post %s", + device.id, + targets.post_id, + ) + try: + await deactivate_push_device(push_device_id=device.id) + except Exception: + logger.exception("Failed to deactivate push device %s", device.id) + _mark_sent(post_id=targets.post_id, push_device_id=device.id) + return "permanent_failed" + except Exception: + logger.exception( + "Transient FCM failure for device %s on group post %s", + device.id, + targets.post_id, + ) + return "transient_failed" + + +async def process_group_post_notification_message(message: Dict[str, Any]) -> None: + receipt_handle = message.get("ReceiptHandle") + body = parse_group_post_notification_message_body(message.get("Body", "")) + if not body: + if receipt_handle: + delete_group_post_notification_message(receipt_handle) + return + + post_id = _parse_uuid(body.get("post_id")) + if not post_id: + logger.error("Invalid post_id in group post notification SQS message: %s", body) + if receipt_handle: + delete_group_post_notification_message(receipt_handle) + return + + if not get_bool("NOTIFICATION_DISPATCH_ENABLED"): + logger.info("Group post notification dispatch disabled; deleting event for %s", post_id) + if receipt_handle: + delete_group_post_notification_message(receipt_handle) + return + + try: + targets = await _fetch_all_targets(post_id) + except HTTPException as exc: + if exc.status_code == 404: + logger.error("Group post not found for notification event %s", post_id) + if receipt_handle: + delete_group_post_notification_message(receipt_handle) + return + raise TransientGroupPostNotificationError(str(exc.detail)) from exc + + devices = [ + device + for recipient in targets.recipients + for device in recipient.push_devices + ] + if not devices: + logger.info("No push devices for group post %s; deleting event", post_id) + if receipt_handle: + delete_group_post_notification_message(receipt_handle) + return + + concurrency = max(get_int("GROUP_POST_NOTIFICATION_SEND_CONCURRENCY"), 1) + semaphore = asyncio.Semaphore(concurrency) + results = await asyncio.gather( + *[ + _send_to_device(targets=targets, device=device, semaphore=semaphore) + for device in devices + ] + ) + + sent = results.count("sent") + skipped = results.count("skipped") + permanent_failed = results.count("permanent_failed") + transient_failed = results.count("transient_failed") + logger.info( + "Group post notification %s processed: sent=%s permanent_failed=%s transient_failed=%s skipped=%s", + post_id, + sent, + permanent_failed, + transient_failed, + skipped, + ) + + if transient_failed > 0: + raise TransientGroupPostNotificationError( + f"Transient failures remain for group post {post_id}" + ) + + if receipt_handle: + delete_group_post_notification_message(receipt_handle) + + +async def run_group_post_notification_sqs_consumer(stop_event: asyncio.Event) -> None: + logger.info("Group post notification SQS consumer started") + while not stop_event.is_set(): + if not is_group_post_notification_sqs_poll_enabled(): + try: + await asyncio.wait_for(stop_event.wait(), timeout=_POLL_IDLE_SECONDS) + except asyncio.TimeoutError: + pass + continue + + try: + messages = await asyncio.to_thread(receive_group_post_notification_messages) + if not messages: + continue + for message in messages: + if stop_event.is_set(): + break + try: + await process_group_post_notification_message(message) + except TransientGroupPostNotificationError: + logger.warning( + "Leaving group post notification SQS message for retry: %s", + message.get("MessageId"), + ) + except Exception: + logger.exception("Unexpected group post notification consumer error") + except Exception: + logger.exception("Group post notification SQS consumer loop error") + try: + await asyncio.wait_for(stop_event.wait(), timeout=_POLL_ERROR_SECONDS) + except asyncio.TimeoutError: + pass + + logger.info("Group post notification SQS consumer stopped") diff --git a/worker_api/notifications/services/push/fcm_client.py b/worker_api/notifications/services/push/fcm_client.py index 6044579..a1cb898 100644 --- a/worker_api/notifications/services/push/fcm_client.py +++ b/worker_api/notifications/services/push/fcm_client.py @@ -59,6 +59,28 @@ def build_chat_notification_data( } +def build_group_post_notification_data( + *, + post_id: UUID, + group_id: UUID, + author_id: UUID, + title: str, + body: str, +) -> dict[str, str]: + """FCM data payloads require string values.""" + return { + "notification_type": "GROUP_POST", + "session_type": "GROUP_POST", + "post_id": str(post_id), + "group_id": str(group_id), + "author_id": str(author_id), + "source_id": str(post_id), + "title": title, + "body": body, + "image_url": "", + } + + async def send_routine_push_notification( *, device_token: str, @@ -146,6 +168,29 @@ async def send_chat_push_notification( ) +async def send_group_post_push_notification( + *, + device_token: str, + post_id: UUID, + group_id: UUID, + author_id: UUID, + title: str, + body: str, +) -> None: + await send_fcm_notification( + device_token=device_token, + title=title, + body=body, + data=build_group_post_notification_data( + post_id=post_id, + group_id=group_id, + author_id=author_id, + title=title, + body=body, + ), + ) + + def build_join_request_notification_data( *, event_type: str, From 8eaac4d2af56efab50cdd86f012f2e260fc8fddd Mon Sep 17 00:00:00 2001 From: tenkus47 Date: Mon, 24 Aug 2026 16:11:19 +0530 Subject: [PATCH 2/2] Add event notification features - Implemented the `fetch_event_notification_targets` function in the backend client to retrieve notification targets for events. - Added schemas for event notifications, including `EventNotificationTargetsResponse` and related models. - Created functions for building and sending event notifications, including `build_event_notification_data` and `send_event_push_notification`. - Introduced unit tests for the new event notification functionalities, ensuring proper response parsing and payload structure. - Updated configuration settings for event notifications in the application. --- tests/notifications/test_backend_client.py | 67 +++++ .../test_event_notification_consumer.py | 220 +++++++++++++++ tests/notifications/test_event_sqs_client.py | 114 ++++++++ tests/notifications/test_fcm_client.py | 51 ++++ worker_api/config.py | 11 + worker_api/db/mongo_database.py | 4 + worker_api/notifications/event_sqs_client.py | 98 +++++++ worker_api/notifications/schemas.py | 24 ++ .../notifications/services/backend_client.py | 17 ++ .../services/event_notification_consumer.py | 257 ++++++++++++++++++ .../notifications/services/push/fcm_client.py | 45 +++ 11 files changed, 908 insertions(+) create mode 100644 tests/notifications/test_event_notification_consumer.py create mode 100644 tests/notifications/test_event_sqs_client.py create mode 100644 worker_api/notifications/event_sqs_client.py create mode 100644 worker_api/notifications/services/event_notification_consumer.py diff --git a/tests/notifications/test_backend_client.py b/tests/notifications/test_backend_client.py index 7be4147..38c25a2 100644 --- a/tests/notifications/test_backend_client.py +++ b/tests/notifications/test_backend_client.py @@ -264,6 +264,73 @@ async def test_defaults_to_first_page(self): assert http_client.get.await_args.kwargs["params"] == {"skip": 0, "limit": 100} +class TestFetchEventNotificationTargets: + @pytest.mark.asyncio + async def test_returns_parsed_targets(self): + event_id = uuid4() + device_id = uuid4() + response = _json_response( + { + "event_id": str(event_id), + "group_id": str(uuid4()), + "author_id": str(uuid4()), + "title": "Sangha", + "body": "Full Moon Meditation", + "recipients": [ + { + "user_id": str(uuid4()), + "push_devices": [ + {"id": str(device_id), "token": "token-1", "platform": "ios"} + ], + } + ], + "skip": 100, + "limit": 50, + "total": 120, + "has_more": False, + } + ) + client_patch, http_client = _patch_async_client(response) + + with client_patch, _patch_config(): + targets = await backend_client.fetch_event_notification_targets( + event_id=event_id, + skip=100, + limit=50, + ) + + assert targets.total == 120 + assert targets.recipients[0].push_devices[0].id == device_id + assert http_client.get.await_args.args[0] == ( + f"http://backend.test/internal/event-notification-targets/{event_id}" + ) + assert http_client.get.await_args.kwargs["params"] == {"skip": 100, "limit": 50} + + @pytest.mark.asyncio + async def test_defaults_to_first_page(self): + event_id = uuid4() + response = _json_response( + { + "event_id": str(event_id), + "group_id": str(uuid4()), + "author_id": str(uuid4()), + "title": "Sangha", + "body": "Full Moon Meditation", + "recipients": [], + "skip": 0, + "limit": 100, + "total": 0, + "has_more": False, + } + ) + client_patch, http_client = _patch_async_client(response) + + with client_patch, _patch_config(): + await backend_client.fetch_event_notification_targets(event_id=event_id) + + assert http_client.get.await_args.kwargs["params"] == {"skip": 0, "limit": 100} + + class TestFetchVerseOfDayNotificationTargets: @pytest.mark.asyncio async def test_returns_parsed_targets(self): diff --git a/tests/notifications/test_event_notification_consumer.py b/tests/notifications/test_event_notification_consumer.py new file mode 100644 index 0000000..38c061f --- /dev/null +++ b/tests/notifications/test_event_notification_consumer.py @@ -0,0 +1,220 @@ +"""Tests for event notification SQS consumer.""" +import json +from unittest.mock import AsyncMock, patch +from uuid import uuid4 + +import pytest + +from worker_api.notifications.schemas import ( + EventNotificationRecipient, + EventNotificationTargetsResponse, + EventPushDeviceTarget, +) +from worker_api.notifications.services.event_notification_consumer import ( + TransientEventNotificationError, + process_event_notification_message, +) +from worker_api.notifications.services.push.fcm_client import PermanentPushTokenError + + +def _targets(*, event_id, devices): + return EventNotificationTargetsResponse( + event_id=event_id, + group_id=uuid4(), + author_id=uuid4(), + title="Sangha", + body="Full Moon Meditation", + recipients=[ + EventNotificationRecipient( + user_id=uuid4(), + push_devices=devices, + ) + ], + skip=0, + limit=100, + total=1, + has_more=False, + ) + + +def _body(event_id): + return json.dumps( + { + "event_type": "EVENT_CREATED", + "version": 1, + "event_id": str(event_id), + } + ) + + +class TestProcessEventNotificationMessage: + @pytest.mark.asyncio + @patch("worker_api.notifications.services.event_notification_consumer.delete_event_notification_message") + async def test_deletes_malformed_message(self, mock_delete): + await process_event_notification_message( + {"ReceiptHandle": "r1", "Body": "not-json"} + ) + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch("worker_api.notifications.services.event_notification_consumer.delete_event_notification_message") + @patch( + "worker_api.notifications.services.event_notification_consumer._fetch_all_targets", + new_callable=AsyncMock, + ) + @patch("worker_api.notifications.services.event_notification_consumer.get_bool", return_value=True) + async def test_deletes_when_event_not_found(self, _get_bool, mock_fetch, mock_delete): + from fastapi import HTTPException + + mock_fetch.side_effect = HTTPException(status_code=404, detail="not found") + event_id = uuid4() + await process_event_notification_message( + {"ReceiptHandle": "r1", "Body": _body(event_id)} + ) + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch("worker_api.notifications.services.event_notification_consumer.delete_event_notification_message") + @patch( + "worker_api.notifications.services.event_notification_consumer.send_event_push_notification", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.event_notification_consumer._fetch_all_targets", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.event_notification_consumer.is_push_configured", + return_value=True, + ) + @patch( + "worker_api.notifications.services.event_notification_consumer._already_sent", + return_value=False, + ) + @patch("worker_api.notifications.services.event_notification_consumer._mark_sent") + @patch("worker_api.notifications.services.event_notification_consumer.get_int", return_value=5) + @patch("worker_api.notifications.services.event_notification_consumer.get_bool", return_value=True) + async def test_sends_and_deletes_on_success( + self, _get_bool, _get_int, mock_mark, _already, _configured, mock_fetch, mock_send, mock_delete, + ): + event_id = uuid4() + device = EventPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(event_id=event_id, devices=[device]) + + await process_event_notification_message( + {"ReceiptHandle": "r1", "Body": _body(event_id)} + ) + + mock_send.assert_awaited_once() + mock_mark.assert_called_once() + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch("worker_api.notifications.services.event_notification_consumer.delete_event_notification_message") + @patch( + "worker_api.notifications.services.event_notification_consumer.deactivate_push_device", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.event_notification_consumer.send_event_push_notification", + new_callable=AsyncMock, + side_effect=PermanentPushTokenError("gone"), + ) + @patch( + "worker_api.notifications.services.event_notification_consumer._fetch_all_targets", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.event_notification_consumer.is_push_configured", + return_value=True, + ) + @patch( + "worker_api.notifications.services.event_notification_consumer._already_sent", + return_value=False, + ) + @patch("worker_api.notifications.services.event_notification_consumer._mark_sent") + @patch("worker_api.notifications.services.event_notification_consumer.get_int", return_value=5) + @patch("worker_api.notifications.services.event_notification_consumer.get_bool", return_value=True) + async def test_permanent_token_deactivates_and_deletes( + self, _get_bool, _get_int, mock_mark, _already, _configured, mock_fetch, mock_send, mock_deactivate, mock_delete, + ): + event_id = uuid4() + device = EventPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(event_id=event_id, devices=[device]) + + await process_event_notification_message( + {"ReceiptHandle": "r1", "Body": _body(event_id)} + ) + + mock_deactivate.assert_awaited_once_with(push_device_id=device.id) + mock_mark.assert_called_once() + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch("worker_api.notifications.services.event_notification_consumer.delete_event_notification_message") + @patch( + "worker_api.notifications.services.event_notification_consumer.send_event_push_notification", + new_callable=AsyncMock, + side_effect=RuntimeError("temporary"), + ) + @patch( + "worker_api.notifications.services.event_notification_consumer._fetch_all_targets", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.event_notification_consumer.is_push_configured", + return_value=True, + ) + @patch( + "worker_api.notifications.services.event_notification_consumer._already_sent", + return_value=False, + ) + @patch("worker_api.notifications.services.event_notification_consumer.get_int", return_value=5) + @patch("worker_api.notifications.services.event_notification_consumer.get_bool", return_value=True) + async def test_transient_failure_leaves_message( + self, _get_bool, _get_int, _already, _configured, mock_fetch, mock_send, mock_delete, + ): + event_id = uuid4() + device = EventPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(event_id=event_id, devices=[device]) + + with pytest.raises(TransientEventNotificationError): + await process_event_notification_message( + {"ReceiptHandle": "r1", "Body": _body(event_id)} + ) + + mock_delete.assert_not_called() + + @pytest.mark.asyncio + @patch("worker_api.notifications.services.event_notification_consumer.delete_event_notification_message") + @patch( + "worker_api.notifications.services.event_notification_consumer.send_event_push_notification", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.event_notification_consumer._fetch_all_targets", + new_callable=AsyncMock, + ) + @patch( + "worker_api.notifications.services.event_notification_consumer.is_push_configured", + return_value=True, + ) + @patch( + "worker_api.notifications.services.event_notification_consumer._already_sent", + return_value=True, + ) + @patch("worker_api.notifications.services.event_notification_consumer.get_int", return_value=5) + @patch("worker_api.notifications.services.event_notification_consumer.get_bool", return_value=True) + async def test_skips_already_sent_devices( + self, _get_bool, _get_int, _already, _configured, mock_fetch, mock_send, mock_delete, + ): + event_id = uuid4() + device = EventPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(event_id=event_id, devices=[device]) + + await process_event_notification_message( + {"ReceiptHandle": "r1", "Body": _body(event_id)} + ) + + mock_send.assert_not_called() + mock_delete.assert_called_once_with("r1") diff --git a/tests/notifications/test_event_sqs_client.py b/tests/notifications/test_event_sqs_client.py new file mode 100644 index 0000000..1f15142 --- /dev/null +++ b/tests/notifications/test_event_sqs_client.py @@ -0,0 +1,114 @@ +"""Tests for event notification SQS client helpers.""" +import json +from unittest.mock import MagicMock, patch +from uuid import uuid4 + +from botocore.exceptions import ClientError + +from worker_api.notifications.event_sqs_client import ( + EVENT_CREATED_EVENT, + EVENT_NOTIFICATION_EVENT_VERSION, + delete_event_notification_message, + get_event_notification_sqs_queue_url, + is_event_notification_sqs_configured, + is_event_notification_sqs_poll_enabled, + parse_event_notification_message_body, + receive_event_notification_messages, +) + + +class TestQueueConfig: + @patch("worker_api.notifications.event_sqs_client.get", return_value=" https://sqs.example/event ") + def test_get_queue_url_strips(self, _get): + assert get_event_notification_sqs_queue_url() == "https://sqs.example/event" + + @patch("worker_api.notifications.event_sqs_client.get_event_notification_sqs_queue_url", return_value="") + def test_is_configured_false(self, _url): + assert is_event_notification_sqs_configured() is False + + @patch("worker_api.notifications.event_sqs_client.is_event_notification_sqs_configured", return_value=True) + @patch("worker_api.notifications.event_sqs_client.get_bool", return_value=True) + def test_poll_enabled(self, _bool, _configured): + assert is_event_notification_sqs_poll_enabled() is True + + +class TestParseBody: + def test_valid_event(self): + event_id = str(uuid4()) + body = parse_event_notification_message_body( + json.dumps( + { + "event_type": EVENT_CREATED_EVENT, + "version": EVENT_NOTIFICATION_EVENT_VERSION, + "event_id": event_id, + } + ) + ) + assert body["event_id"] == event_id + + def test_rejects_invalid_json(self): + assert parse_event_notification_message_body("not-json") is None + + def test_rejects_wrong_event_type(self): + assert parse_event_notification_message_body( + json.dumps({"event_type": "OTHER", "version": 1, "event_id": str(uuid4())}) + ) is None + + def test_rejects_wrong_version(self): + assert parse_event_notification_message_body( + json.dumps( + { + "event_type": EVENT_CREATED_EVENT, + "version": 99, + "event_id": str(uuid4()), + } + ) + ) is None + + def test_rejects_missing_event_id(self): + assert parse_event_notification_message_body( + json.dumps({"event_type": EVENT_CREATED_EVENT, "version": 1}) + ) is None + + +class TestReceiveAndDelete: + @patch("worker_api.notifications.event_sqs_client.get_event_notification_sqs_queue_url", return_value="") + def test_receive_empty_without_queue(self, _url): + assert receive_event_notification_messages() == [] + + @patch("worker_api.notifications.event_sqs_client.get_int", side_effect=lambda key: 1) + @patch("worker_api.notifications.event_sqs_client._get_sqs_client") + @patch( + "worker_api.notifications.event_sqs_client.get_event_notification_sqs_queue_url", + return_value="https://sqs.example/event", + ) + def test_receive_messages(self, _url, mock_get_client, _get_int): + client = MagicMock() + client.receive_message.return_value = {"Messages": [{"MessageId": "1"}]} + mock_get_client.return_value = client + assert receive_event_notification_messages() == [{"MessageId": "1"}] + + @patch("worker_api.notifications.event_sqs_client._get_sqs_client") + @patch( + "worker_api.notifications.event_sqs_client.get_event_notification_sqs_queue_url", + return_value="https://sqs.example/event", + ) + def test_delete_message(self, _url, mock_get_client): + client = MagicMock() + mock_get_client.return_value = client + delete_event_notification_message("receipt") + client.delete_message.assert_called_once() + + @patch("worker_api.notifications.event_sqs_client._get_sqs_client") + @patch( + "worker_api.notifications.event_sqs_client.get_event_notification_sqs_queue_url", + return_value="https://sqs.example/event", + ) + def test_delete_handles_client_error(self, _url, mock_get_client): + client = MagicMock() + client.delete_message.side_effect = ClientError( + {"Error": {"Code": "500", "Message": "fail"}}, + "DeleteMessage", + ) + mock_get_client.return_value = client + delete_event_notification_message("receipt") diff --git a/tests/notifications/test_fcm_client.py b/tests/notifications/test_fcm_client.py index bcc1d91..67c05b7 100644 --- a/tests/notifications/test_fcm_client.py +++ b/tests/notifications/test_fcm_client.py @@ -5,9 +5,11 @@ from worker_api.notifications.services.push.fcm_client import ( build_chat_notification_data, + build_event_notification_data, build_group_post_notification_data, build_routine_notification_data, send_chat_push_notification, + send_event_push_notification, send_fcm_notification, send_group_post_push_notification, send_routine_push_notification, @@ -116,6 +118,31 @@ def test_includes_group_post_routing_fields(self): } +class TestBuildEventNotificationData: + def test_includes_event_routing_fields(self): + event_id = uuid4() + group_id = uuid4() + author_id = uuid4() + data = build_event_notification_data( + event_id=event_id, + group_id=group_id, + author_id=author_id, + title="Sangha", + body="Full Moon Meditation", + ) + assert data == { + "notification_type": "EVENT", + "session_type": "EVENT", + "event_id": str(event_id), + "group_id": str(group_id), + "author_id": str(author_id), + "source_id": str(event_id), + "title": "Sangha", + "body": "Full Moon Meditation", + "image_url": "", + } + + class TestSendFcmNotification: @pytest.mark.asyncio @patch("worker_api.notifications.services.push.fcm_client.messaging.send") @@ -235,3 +262,27 @@ async def test_delegates_with_group_post_payload(self, mock_send): assert kwargs["data"]["notification_type"] == "GROUP_POST" assert kwargs["data"]["session_type"] == "GROUP_POST" assert kwargs["data"]["post_id"] == str(post_id) + + +class TestSendEventPushNotification: + @pytest.mark.asyncio + @patch("worker_api.notifications.services.push.fcm_client.send_fcm_notification") + async def test_delegates_with_event_payload(self, mock_send): + event_id = uuid4() + group_id = uuid4() + author_id = uuid4() + await send_event_push_notification( + device_token="device-token", + event_id=event_id, + group_id=group_id, + author_id=author_id, + title="Sangha", + body="Full Moon Meditation", + ) + mock_send.assert_awaited_once() + kwargs = mock_send.await_args.kwargs + assert kwargs["title"] == "Sangha" + assert kwargs["body"] == "Full Moon Meditation" + assert kwargs["data"]["notification_type"] == "EVENT" + assert kwargs["data"]["session_type"] == "EVENT" + assert kwargs["data"]["event_id"] == str(event_id) diff --git a/worker_api/config.py b/worker_api/config.py index f1d8f9a..d92b500 100644 --- a/worker_api/config.py +++ b/worker_api/config.py @@ -69,6 +69,17 @@ GROUP_POST_NOTIFICATION_IDEMPOTENCY_TTL_SECONDS=86400, GROUP_POST_NOTIFICATION_IDEMPOTENCY_KEY_PREFIX="worker:group-post-notifications:sent:", + # Event notification SQS consumer (backend producer → worker consumer) + EVENT_NOTIFICATION_SQS_QUEUE_URL="", + EVENT_NOTIFICATION_SQS_WAIT_TIME_SECONDS=20, + EVENT_NOTIFICATION_SQS_VISIBILITY_TIMEOUT_SECONDS=300, + EVENT_NOTIFICATION_SQS_MAX_MESSAGES=5, + EVENT_NOTIFICATION_SQS_POLL_ENABLED="true", + EVENT_NOTIFICATION_SEND_CONCURRENCY=10, + EVENT_NOTIFICATION_TARGET_PAGE_SIZE=100, + EVENT_NOTIFICATION_IDEMPOTENCY_TTL_SECONDS=86400, + EVENT_NOTIFICATION_IDEMPOTENCY_KEY_PREFIX="worker:event-notifications:sent:", + # Notification dispatch (Cloud Scheduler -> worker) NOTIFICATION_DISPATCH_SECRET_TOKEN="Dispatch", NOTIFICATION_DISPATCH_BATCH_SIZE=100, diff --git a/worker_api/db/mongo_database.py b/worker_api/db/mongo_database.py index 7fc55a8..a1d0abb 100644 --- a/worker_api/db/mongo_database.py +++ b/worker_api/db/mongo_database.py @@ -11,6 +11,9 @@ from worker_api.notifications.services.chat_notification_consumer import ( run_chat_notification_sqs_consumer, ) +from worker_api.notifications.services.event_notification_consumer import ( + run_event_notification_sqs_consumer, +) from worker_api.notifications.services.group_post_notification_consumer import ( run_group_post_notification_sqs_consumer, ) @@ -46,6 +49,7 @@ async def lifespan(api: FastAPI): asyncio.create_task(run_chat_notification_sqs_consumer(stop_event)), asyncio.create_task(run_join_request_notification_sqs_consumer(stop_event)), asyncio.create_task(run_group_post_notification_sqs_consumer(stop_event)), + asyncio.create_task(run_event_notification_sqs_consumer(stop_event)), ] yield diff --git a/worker_api/notifications/event_sqs_client.py b/worker_api/notifications/event_sqs_client.py new file mode 100644 index 0000000..f02838b --- /dev/null +++ b/worker_api/notifications/event_sqs_client.py @@ -0,0 +1,98 @@ +import json +import logging +from typing import Any, Dict, List, Optional + +import boto3 +from botocore.exceptions import ClientError + +from worker_api.config import get, get_bool, get_int + +logger = logging.getLogger(__name__) + +_sqs_client = None + +EVENT_CREATED_EVENT = "EVENT_CREATED" +EVENT_NOTIFICATION_EVENT_VERSION = 1 + + +def _get_sqs_client(): + global _sqs_client + if _sqs_client is None: + _sqs_client = boto3.client( + "sqs", + aws_access_key_id=get("AWS_ACCESS_KEY"), + aws_secret_access_key=get("AWS_SECRET_KEY"), + region_name=get("AWS_REGION"), + ) + return _sqs_client + + +def get_event_notification_sqs_queue_url() -> str: + return get("EVENT_NOTIFICATION_SQS_QUEUE_URL").strip() + + +def is_event_notification_sqs_configured() -> bool: + return bool(get_event_notification_sqs_queue_url()) + + +def is_event_notification_sqs_poll_enabled() -> bool: + return get_bool("EVENT_NOTIFICATION_SQS_POLL_ENABLED") and is_event_notification_sqs_configured() + + +def receive_event_notification_messages() -> List[Dict[str, Any]]: + queue_url = get_event_notification_sqs_queue_url() + if not queue_url: + return [] + + try: + response = _get_sqs_client().receive_message( + QueueUrl=queue_url, + MaxNumberOfMessages=get_int("EVENT_NOTIFICATION_SQS_MAX_MESSAGES"), + WaitTimeSeconds=get_int("EVENT_NOTIFICATION_SQS_WAIT_TIME_SECONDS"), + VisibilityTimeout=get_int("EVENT_NOTIFICATION_SQS_VISIBILITY_TIMEOUT_SECONDS"), + MessageAttributeNames=["All"], + ) + return response.get("Messages", []) + except ClientError as e: + logger.error("Failed to receive event notification SQS messages: %s", e) + return [] + + +def delete_event_notification_message(receipt_handle: str) -> None: + queue_url = get_event_notification_sqs_queue_url() + if not queue_url or not receipt_handle: + return + + try: + _get_sqs_client().delete_message( + QueueUrl=queue_url, + ReceiptHandle=receipt_handle, + ) + except ClientError as e: + logger.error("Failed to delete event notification SQS message: %s", e) + + +def parse_event_notification_message_body(raw_body: str) -> Optional[Dict[str, Any]]: + try: + body = json.loads(raw_body) + except (TypeError, json.JSONDecodeError) as e: + logger.error("Invalid event notification SQS message body: %s", e) + return None + + if not isinstance(body, dict): + logger.error("Event notification SQS message body is not an object: %s", body) + return None + + event_type = body.get("event_type") + version = body.get("version") + event_id = body.get("event_id") + if event_type != EVENT_CREATED_EVENT: + logger.error("Unsupported event notification event_type: %s", event_type) + return None + if version != EVENT_NOTIFICATION_EVENT_VERSION: + logger.error("Unsupported event notification event version: %s", version) + return None + if not event_id: + logger.error("Event notification SQS message missing event_id: %s", body) + return None + return body diff --git a/worker_api/notifications/schemas.py b/worker_api/notifications/schemas.py index 348611d..1220aaf 100644 --- a/worker_api/notifications/schemas.py +++ b/worker_api/notifications/schemas.py @@ -217,6 +217,30 @@ class GroupPostNotificationTargetsResponse(BaseModel): has_more: bool +class EventPushDeviceTarget(BaseModel): + id: UUID + token: str + platform: str + + +class EventNotificationRecipient(BaseModel): + user_id: UUID + push_devices: list[EventPushDeviceTarget] + + +class EventNotificationTargetsResponse(BaseModel): + event_id: UUID + group_id: UUID + author_id: UUID + title: str + body: str + recipients: list[EventNotificationRecipient] + skip: int + limit: int + total: int + has_more: bool + + class JoinRequestPushDeviceTarget(BaseModel): id: UUID token: str diff --git a/worker_api/notifications/services/backend_client.py b/worker_api/notifications/services/backend_client.py index 63821f5..b82ef1b 100644 --- a/worker_api/notifications/services/backend_client.py +++ b/worker_api/notifications/services/backend_client.py @@ -6,6 +6,7 @@ from worker_api.notifications.schemas import ( ChatNotificationTargetsResponse, DeactivatePushDeviceResponse, + EventNotificationTargetsResponse, GroupPostNotificationTargetsResponse, JoinRequestNotificationTargetsResponse, NotificationContent, @@ -111,6 +112,22 @@ async def fetch_group_post_notification_targets( return GroupPostNotificationTargetsResponse.model_validate(response.json()) +async def fetch_event_notification_targets( + *, + event_id: UUID, + skip: int = 0, + limit: int = 100, +) -> EventNotificationTargetsResponse: + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.get( + f"{_backend_url()}/internal/event-notification-targets/{event_id}", + params={"skip": skip, "limit": limit}, + headers=_backend_headers(), + ) + response.raise_for_status() + return EventNotificationTargetsResponse.model_validate(response.json()) + + async def deactivate_push_device(*, push_device_id: UUID) -> DeactivatePushDeviceResponse: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( diff --git a/worker_api/notifications/services/event_notification_consumer.py b/worker_api/notifications/services/event_notification_consumer.py new file mode 100644 index 0000000..05910a9 --- /dev/null +++ b/worker_api/notifications/services/event_notification_consumer.py @@ -0,0 +1,257 @@ +import asyncio +import logging +from typing import Any, Dict, Optional +from uuid import UUID + +import httpx +import redis +from fastapi import HTTPException + +from worker_api.config import get, get_bool, get_int +from worker_api.notifications.event_sqs_client import ( + delete_event_notification_message, + is_event_notification_sqs_poll_enabled, + parse_event_notification_message_body, + receive_event_notification_messages, +) +from worker_api.notifications.schemas import ( + EventNotificationTargetsResponse, + EventPushDeviceTarget, +) +from worker_api.notifications.services.backend_client import ( + deactivate_push_device, + fetch_event_notification_targets, +) +from worker_api.notifications.services.push.config_loader import is_push_configured +from worker_api.notifications.services.push.fcm_client import ( + PermanentPushTokenError, + send_event_push_notification, +) + +logger = logging.getLogger(__name__) + +_POLL_IDLE_SECONDS = 5 +_POLL_ERROR_SECONDS = 10 +_redis_client: redis.Redis | None = None + + +class TransientEventNotificationError(Exception): + """Raised when processing should leave the SQS message for retry.""" + + +def _get_redis_client() -> redis.Redis: + global _redis_client + if _redis_client is None: + _redis_client = redis.Redis.from_url(get("CACHE_CONNECTION_STRING")) + return _redis_client + + +def _idempotency_key(*, event_id: UUID, push_device_id: UUID) -> str: + prefix = get("EVENT_NOTIFICATION_IDEMPOTENCY_KEY_PREFIX") + return f"{prefix}{event_id}:{push_device_id}" + + +def _already_sent(*, event_id: UUID, push_device_id: UUID) -> bool: + client = _get_redis_client() + return bool(client.exists(_idempotency_key(event_id=event_id, push_device_id=push_device_id))) + + +def _mark_sent(*, event_id: UUID, push_device_id: UUID) -> None: + client = _get_redis_client() + client.setex( + _idempotency_key(event_id=event_id, push_device_id=push_device_id), + get_int("EVENT_NOTIFICATION_IDEMPOTENCY_TTL_SECONDS"), + "1", + ) + + +def _parse_uuid(value: Any) -> Optional[UUID]: + if value is None or value == "": + return None + return UUID(str(value)) + + +async def _fetch_all_targets(event_id: UUID) -> EventNotificationTargetsResponse: + page_size = max(get_int("EVENT_NOTIFICATION_TARGET_PAGE_SIZE"), 1) + skip = 0 + first_page: EventNotificationTargetsResponse | None = None + all_recipients = [] + + while True: + try: + page = await fetch_event_notification_targets( + event_id=event_id, + skip=skip, + limit=page_size, + ) + except httpx.HTTPStatusError as exc: + if exc.response.status_code == 404: + raise HTTPException(status_code=404, detail="Event not found") from exc + raise TransientEventNotificationError(str(exc)) from exc + except httpx.HTTPError as exc: + raise TransientEventNotificationError(str(exc)) from exc + + if first_page is None: + first_page = page + all_recipients.extend(page.recipients) + if not page.has_more: + break + skip += page.limit + + assert first_page is not None + return first_page.model_copy(update={"recipients": all_recipients, "has_more": False}) + + +async def _send_to_device( + *, + targets: EventNotificationTargetsResponse, + device: EventPushDeviceTarget, + semaphore: asyncio.Semaphore, +) -> str: + """Return sent | skipped | permanent_failed | transient_failed.""" + async with semaphore: + if not is_push_configured(device.platform): + return "skipped" + + if _already_sent(event_id=targets.event_id, push_device_id=device.id): + return "skipped" + + try: + await send_event_push_notification( + device_token=device.token, + event_id=targets.event_id, + group_id=targets.group_id, + author_id=targets.author_id, + title=targets.title, + body=targets.body, + ) + _mark_sent(event_id=targets.event_id, push_device_id=device.id) + return "sent" + except PermanentPushTokenError: + logger.warning( + "Deactivating permanently invalid push device %s for event %s", + device.id, + targets.event_id, + ) + try: + await deactivate_push_device(push_device_id=device.id) + except Exception: + logger.exception("Failed to deactivate push device %s", device.id) + _mark_sent(event_id=targets.event_id, push_device_id=device.id) + return "permanent_failed" + except Exception: + logger.exception( + "Transient FCM failure for device %s on event %s", + device.id, + targets.event_id, + ) + return "transient_failed" + + +async def process_event_notification_message(message: Dict[str, Any]) -> None: + receipt_handle = message.get("ReceiptHandle") + body = parse_event_notification_message_body(message.get("Body", "")) + if not body: + if receipt_handle: + delete_event_notification_message(receipt_handle) + return + + event_id = _parse_uuid(body.get("event_id")) + if not event_id: + logger.error("Invalid event_id in event notification SQS message: %s", body) + if receipt_handle: + delete_event_notification_message(receipt_handle) + return + + if not get_bool("NOTIFICATION_DISPATCH_ENABLED"): + logger.info("Event notification dispatch disabled; deleting event for %s", event_id) + if receipt_handle: + delete_event_notification_message(receipt_handle) + return + + try: + targets = await _fetch_all_targets(event_id) + except HTTPException as exc: + if exc.status_code == 404: + logger.error("Event not found for notification event %s", event_id) + if receipt_handle: + delete_event_notification_message(receipt_handle) + return + raise TransientEventNotificationError(str(exc.detail)) from exc + + devices = [ + device + for recipient in targets.recipients + for device in recipient.push_devices + ] + if not devices: + logger.info("No push devices for event %s; deleting event", event_id) + if receipt_handle: + delete_event_notification_message(receipt_handle) + return + + concurrency = max(get_int("EVENT_NOTIFICATION_SEND_CONCURRENCY"), 1) + semaphore = asyncio.Semaphore(concurrency) + results = await asyncio.gather( + *[ + _send_to_device(targets=targets, device=device, semaphore=semaphore) + for device in devices + ] + ) + + sent = results.count("sent") + skipped = results.count("skipped") + permanent_failed = results.count("permanent_failed") + transient_failed = results.count("transient_failed") + logger.info( + "Event notification %s processed: sent=%s permanent_failed=%s transient_failed=%s skipped=%s", + event_id, + sent, + permanent_failed, + transient_failed, + skipped, + ) + + if transient_failed > 0: + raise TransientEventNotificationError( + f"Transient failures remain for event {event_id}" + ) + + if receipt_handle: + delete_event_notification_message(receipt_handle) + + +async def run_event_notification_sqs_consumer(stop_event: asyncio.Event) -> None: + logger.info("Event notification SQS consumer started") + while not stop_event.is_set(): + if not is_event_notification_sqs_poll_enabled(): + try: + await asyncio.wait_for(stop_event.wait(), timeout=_POLL_IDLE_SECONDS) + except asyncio.TimeoutError: + pass + continue + + try: + messages = await asyncio.to_thread(receive_event_notification_messages) + if not messages: + continue + for message in messages: + if stop_event.is_set(): + break + try: + await process_event_notification_message(message) + except TransientEventNotificationError: + logger.warning( + "Leaving event notification SQS message for retry: %s", + message.get("MessageId"), + ) + except Exception: + logger.exception("Unexpected event notification consumer error") + except Exception: + logger.exception("Event notification SQS consumer loop error") + try: + await asyncio.wait_for(stop_event.wait(), timeout=_POLL_ERROR_SECONDS) + except asyncio.TimeoutError: + pass + + logger.info("Event notification SQS consumer stopped") diff --git a/worker_api/notifications/services/push/fcm_client.py b/worker_api/notifications/services/push/fcm_client.py index a1cb898..73a94c3 100644 --- a/worker_api/notifications/services/push/fcm_client.py +++ b/worker_api/notifications/services/push/fcm_client.py @@ -81,6 +81,28 @@ def build_group_post_notification_data( } +def build_event_notification_data( + *, + event_id: UUID, + group_id: UUID, + author_id: UUID, + title: str, + body: str, +) -> dict[str, str]: + """FCM data payloads require string values.""" + return { + "notification_type": "EVENT", + "session_type": "EVENT", + "event_id": str(event_id), + "group_id": str(group_id), + "author_id": str(author_id), + "source_id": str(event_id), + "title": title, + "body": body, + "image_url": "", + } + + async def send_routine_push_notification( *, device_token: str, @@ -191,6 +213,29 @@ async def send_group_post_push_notification( ) +async def send_event_push_notification( + *, + device_token: str, + event_id: UUID, + group_id: UUID, + author_id: UUID, + title: str, + body: str, +) -> None: + await send_fcm_notification( + device_token=device_token, + title=title, + body=body, + data=build_event_notification_data( + event_id=event_id, + group_id=group_id, + author_id=author_id, + title=title, + body=body, + ), + ) + + def build_join_request_notification_data( *, event_type: str,