diff --git a/docs/notification-format.md b/docs/notification-format.md index cc8d6d0..9802b3d 100644 --- a/docs/notification-format.md +++ b/docs/notification-format.md @@ -61,6 +61,7 @@ Routine notifications are sent only for **`PLAN`** and **`SERIES`** sessions. | `PLAN` | Plan UUID | | `SERIES` | Series UUID | | `CHAT` | Chat room UUID | +| `GROUP` | Group UUID | | `VERSE_OF_DAY` | None (`source_id` field not present) | Plan reminder dispatches (enrollment API) always use `session_type: "PLAN"`. @@ -70,6 +71,11 @@ Chat message pushes use `session_type: "CHAT"` plus dedicated routing fields Verse-of-the-day pushes use `session_type: "VERSE_OF_DAY"` plus `notification_type: "VERSE_OF_DAY"`. +Group join-request pushes use `session_type: "GROUP"` with `source_id` set to the +group UUID, plus `notification_type: "JOIN_REQUEST_CREATED"` or +`"JOIN_REQUEST_DECIDED"` and the routing fields `join_request_id`, `group_id`, +and `status`. + ## Default content When no custom content is configured: @@ -337,6 +343,54 @@ When the day notification uses `image_type = CUSTOM`, `image_url` points to the } ``` +### Group join request (moderators notified) + +```json +{ + "notification": { + "title": "Morning Sangha", + "body": "Tenzin Tib asked to join Morning Sangha" + }, + "data": { + "notification_type": "JOIN_REQUEST_CREATED", + "session_type": "GROUP", + "join_request_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", + "group_id": "f4a5b6c7-d8e9-0123-def0-234567890123", + "status": "PENDING", + "source_id": "f4a5b6c7-d8e9-0123-def0-234567890123", + "title": "Morning Sangha", + "body": "Tenzin Tib asked to join Morning Sangha", + "image_url": "" + } +} +``` + +### Group join request decided (requester notified) + +```json +{ + "notification": { + "title": "Morning Sangha", + "body": "You've joined Morning Sangha" + }, + "data": { + "notification_type": "JOIN_REQUEST_DECIDED", + "session_type": "GROUP", + "join_request_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", + "group_id": "f4a5b6c7-d8e9-0123-def0-234567890123", + "status": "APPROVED", + "source_id": "f4a5b6c7-d8e9-0123-def0-234567890123", + "title": "Morning Sangha", + "body": "You've joined Morning Sangha", + "image_url": "" + } +} +``` + +`status` is `APPROVED` or `REJECTED` for `JOIN_REQUEST_DECIDED`, and `PENDING` +for `JOIN_REQUEST_CREATED`. `title` and `body` are composed by the backend; the +worker forwards them unchanged. + ## Chat notification delivery Chat pushes are event-driven: @@ -364,6 +418,37 @@ Configure: Attach a dead-letter queue (DLQ) to the chat notification SQS queue for poison messages. +## Join request notification delivery + +Group join-request pushes are event-driven and use their own SQS queue, separate +from the chat queue: + +1. A user asks to join a private group; a Studio moderator approves or rejects it. +2. Backend enqueues `{ "event_type": "JOIN_REQUEST_CREATED" | "JOIN_REQUEST_DECIDED", "version": 1, "join_request_id": "..." }` to `JOIN_REQUEST_NOTIFICATION_SQS_QUEUE_URL`. +3. Worker consumes the event, paginates `GET /internal/join-request-notification-targets/{join_request_id}`, and sends FCM. +4. Permanently invalid tokens are deactivated via `POST /internal/push-devices/deactivate`. + +**Recipients:** + +| Event | Recipients | +|-------|------------| +| `JOIN_REQUEST_CREATED` | The group's moderators | +| `JOIN_REQUEST_DECIDED` | The requesting user only | + +Recipients with no registered push device are counted in `total` but omitted from +`recipients`; the worker treats that as a no-op and deletes the event. + +Configure: + +| Variable | Purpose | +|----------|---------| +| `JOIN_REQUEST_NOTIFICATION_SQS_QUEUE_URL` | Dedicated SQS queue (backend producer, worker consumer) | +| `JOIN_REQUEST_NOTIFICATION_SQS_POLL_ENABLED` | Worker poll kill switch (`true`/`false`) | +| `JOIN_REQUEST_NOTIFICATION_SEND_CONCURRENCY` | Max concurrent FCM sends per event | +| `JOIN_REQUEST_NOTIFICATION_IDEMPOTENCY_TTL_SECONDS` | Redis TTL for `join_request_id + event_type + push_device_id` dedupe | + +Attach a dead-letter queue (DLQ) to the join request notification SQS queue for poison messages. + ## Preview API vs push payload The backend preview endpoint (`GET /internal/routine-notification-targets`) returns additional server-side fields that are **not** sent to the device: @@ -403,4 +488,5 @@ When the user taps a notification: | `PLAN` | Plan detail / day view | | `SERIES` | Series player | | `CHAT` | Chat room / DM thread using `room_id` (`chat_kind` + optional `group_id`) | +| `GROUP` | Group screen using `group_id`; for `JOIN_REQUEST_CREATED` open the group's pending-requests view | | `VERSE_OF_DAY` | App home / verse-of-day screen (no linked entity) | diff --git a/tests/notifications/test_join_request_notification_consumer.py b/tests/notifications/test_join_request_notification_consumer.py new file mode 100644 index 0000000..0d5565b --- /dev/null +++ b/tests/notifications/test_join_request_notification_consumer.py @@ -0,0 +1,414 @@ +"""Tests for join request notification SQS consumer.""" +import json +from unittest.mock import AsyncMock, patch +from uuid import uuid4 + +import pytest + +from worker_api.notifications.schemas import ( + JoinRequestNotificationRecipient, + JoinRequestNotificationTargetsResponse, + JoinRequestPushDeviceTarget, +) +from worker_api.notifications.services.join_request_notification_consumer import ( + TransientJoinRequestNotificationError, + _idempotency_key, + process_join_request_notification_message, +) +from worker_api.notifications.services.push.fcm_client import PermanentPushTokenError + +CONSUMER = "worker_api.notifications.services.join_request_notification_consumer" + + +def _targets( + *, + join_request_id, + devices, + event_type="JOIN_REQUEST_CREATED", + status="PENDING", + recipients=None, +): + if recipients is None: + recipients = [ + JoinRequestNotificationRecipient(user_id=uuid4(), push_devices=devices) + ] + return JoinRequestNotificationTargetsResponse( + join_request_id=join_request_id, + group_id=uuid4(), + event_type=event_type, + status=status, + group_name="Morning Sangha", + requester_name="Tenzin Tib", + title="Morning Sangha", + body="Tenzin Tib asked to join Morning Sangha", + recipients=recipients, + skip=0, + limit=100, + total=1, + has_more=False, + ) + + +def _sqs_message(*, join_request_id, event_type="JOIN_REQUEST_CREATED", receipt="r1"): + return { + "ReceiptHandle": receipt, + "Body": json.dumps( + { + "event_type": event_type, + "version": 1, + "join_request_id": str(join_request_id), + } + ), + } + + +class TestProcessJoinRequestNotificationMessage: + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + async def test_deletes_malformed_message(self, mock_delete): + await process_join_request_notification_message( + {"ReceiptHandle": "r1", "Body": "not-json"} + ) + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + async def test_deletes_unknown_event_type(self, mock_delete): + await process_join_request_notification_message( + { + "ReceiptHandle": "r1", + "Body": json.dumps( + { + "event_type": "SOMETHING_ELSE", + "version": 1, + "join_request_id": str(uuid4()), + } + ), + } + ) + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + async def test_deletes_wrong_version(self, mock_delete): + await process_join_request_notification_message( + { + "ReceiptHandle": "r1", + "Body": json.dumps( + { + "event_type": "JOIN_REQUEST_CREATED", + "version": 2, + "join_request_id": str(uuid4()), + } + ), + } + ) + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + @patch(f"{CONSUMER}._fetch_all_targets", new_callable=AsyncMock) + @patch(f"{CONSUMER}.get_bool", return_value=False) + async def test_deletes_when_dispatch_disabled(self, _get_bool, mock_fetch, mock_delete): + await process_join_request_notification_message( + _sqs_message(join_request_id=uuid4()) + ) + mock_fetch.assert_not_awaited() + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + @patch(f"{CONSUMER}._fetch_all_targets", new_callable=AsyncMock) + @patch(f"{CONSUMER}.get_bool", return_value=True) + async def test_deletes_when_join_request_not_found(self, _get_bool, mock_fetch, mock_delete): + from fastapi import HTTPException + + mock_fetch.side_effect = HTTPException(status_code=404, detail="not found") + await process_join_request_notification_message( + _sqs_message(join_request_id=uuid4()) + ) + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + @patch(f"{CONSUMER}.send_join_request_push_notification", new_callable=AsyncMock) + @patch(f"{CONSUMER}._fetch_all_targets", new_callable=AsyncMock) + @patch(f"{CONSUMER}.get_int", return_value=5) + @patch(f"{CONSUMER}.get_bool", return_value=True) + async def test_deletes_when_no_recipients( + self, _get_bool, _get_int, mock_fetch, mock_send, mock_delete + ): + join_request_id = uuid4() + mock_fetch.return_value = _targets( + join_request_id=join_request_id, devices=[], recipients=[] + ) + + await process_join_request_notification_message( + _sqs_message(join_request_id=join_request_id) + ) + + mock_send.assert_not_called() + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + @patch(f"{CONSUMER}.send_join_request_push_notification", new_callable=AsyncMock) + @patch(f"{CONSUMER}._fetch_all_targets", new_callable=AsyncMock) + @patch(f"{CONSUMER}.get_int", return_value=5) + @patch(f"{CONSUMER}.get_bool", return_value=True) + async def test_deletes_when_recipient_has_no_devices( + self, _get_bool, _get_int, mock_fetch, mock_send, mock_delete + ): + join_request_id = uuid4() + mock_fetch.return_value = _targets(join_request_id=join_request_id, devices=[]) + + await process_join_request_notification_message( + _sqs_message(join_request_id=join_request_id) + ) + + mock_send.assert_not_called() + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + @patch(f"{CONSUMER}.send_join_request_push_notification", new_callable=AsyncMock) + @patch(f"{CONSUMER}._fetch_all_targets", new_callable=AsyncMock) + @patch(f"{CONSUMER}.is_push_configured", return_value=True) + @patch(f"{CONSUMER}._already_sent", return_value=False) + @patch(f"{CONSUMER}._mark_sent") + @patch(f"{CONSUMER}.get_int", return_value=5) + @patch(f"{CONSUMER}.get_bool", return_value=True) + async def test_sends_created_event_and_deletes( + self, + _get_bool, + _get_int, + mock_mark, + _already, + _configured, + mock_fetch, + mock_send, + mock_delete, + ): + join_request_id = uuid4() + device = JoinRequestPushDeviceTarget(id=uuid4(), token="tok", platform="android") + targets = _targets( + join_request_id=join_request_id, + devices=[device], + event_type="JOIN_REQUEST_CREATED", + status="PENDING", + ) + mock_fetch.return_value = targets + + await process_join_request_notification_message( + _sqs_message(join_request_id=join_request_id, event_type="JOIN_REQUEST_CREATED") + ) + + mock_send.assert_awaited_once_with( + device_token="tok", + event_type="JOIN_REQUEST_CREATED", + join_request_id=join_request_id, + group_id=targets.group_id, + status="PENDING", + title=targets.title, + body=targets.body, + ) + mock_mark.assert_called_once() + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + @patch(f"{CONSUMER}.send_join_request_push_notification", new_callable=AsyncMock) + @patch(f"{CONSUMER}._fetch_all_targets", new_callable=AsyncMock) + @patch(f"{CONSUMER}.is_push_configured", return_value=True) + @patch(f"{CONSUMER}._already_sent", return_value=False) + @patch(f"{CONSUMER}._mark_sent") + @patch(f"{CONSUMER}.get_int", return_value=5) + @patch(f"{CONSUMER}.get_bool", return_value=True) + async def test_sends_decided_event_and_deletes( + self, + _get_bool, + _get_int, + mock_mark, + _already, + _configured, + mock_fetch, + mock_send, + mock_delete, + ): + join_request_id = uuid4() + device = JoinRequestPushDeviceTarget(id=uuid4(), token="tok", platform="ios") + targets = _targets( + join_request_id=join_request_id, + devices=[device], + event_type="JOIN_REQUEST_DECIDED", + status="APPROVED", + ) + mock_fetch.return_value = targets + + await process_join_request_notification_message( + _sqs_message(join_request_id=join_request_id, event_type="JOIN_REQUEST_DECIDED") + ) + + assert mock_send.await_args.kwargs["event_type"] == "JOIN_REQUEST_DECIDED" + assert mock_send.await_args.kwargs["status"] == "APPROVED" + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + @patch(f"{CONSUMER}.deactivate_push_device", new_callable=AsyncMock) + @patch( + f"{CONSUMER}.send_join_request_push_notification", + new_callable=AsyncMock, + side_effect=PermanentPushTokenError("gone"), + ) + @patch(f"{CONSUMER}._fetch_all_targets", new_callable=AsyncMock) + @patch(f"{CONSUMER}.is_push_configured", return_value=True) + @patch(f"{CONSUMER}._already_sent", return_value=False) + @patch(f"{CONSUMER}._mark_sent") + @patch(f"{CONSUMER}.get_int", return_value=5) + @patch(f"{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, + ): + join_request_id = uuid4() + device = JoinRequestPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(join_request_id=join_request_id, devices=[device]) + + await process_join_request_notification_message( + _sqs_message(join_request_id=join_request_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(f"{CONSUMER}.delete_join_request_notification_message") + @patch( + f"{CONSUMER}.send_join_request_push_notification", + new_callable=AsyncMock, + side_effect=RuntimeError("temporary"), + ) + @patch(f"{CONSUMER}._fetch_all_targets", new_callable=AsyncMock) + @patch(f"{CONSUMER}.is_push_configured", return_value=True) + @patch(f"{CONSUMER}._already_sent", return_value=False) + @patch(f"{CONSUMER}.get_int", return_value=5) + @patch(f"{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, + ): + join_request_id = uuid4() + device = JoinRequestPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(join_request_id=join_request_id, devices=[device]) + + with pytest.raises(TransientJoinRequestNotificationError): + await process_join_request_notification_message( + _sqs_message(join_request_id=join_request_id) + ) + + mock_delete.assert_not_called() + + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + @patch(f"{CONSUMER}.send_join_request_push_notification", new_callable=AsyncMock) + @patch(f"{CONSUMER}._fetch_all_targets", new_callable=AsyncMock) + @patch(f"{CONSUMER}.is_push_configured", return_value=True) + @patch(f"{CONSUMER}._already_sent", return_value=True) + @patch(f"{CONSUMER}.get_int", return_value=5) + @patch(f"{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, + ): + join_request_id = uuid4() + device = JoinRequestPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(join_request_id=join_request_id, devices=[device]) + + await process_join_request_notification_message( + _sqs_message(join_request_id=join_request_id) + ) + + mock_send.assert_not_called() + mock_delete.assert_called_once_with("r1") + + @pytest.mark.asyncio + @patch(f"{CONSUMER}.delete_join_request_notification_message") + @patch(f"{CONSUMER}.send_join_request_push_notification", new_callable=AsyncMock) + @patch(f"{CONSUMER}._fetch_all_targets", new_callable=AsyncMock) + @patch(f"{CONSUMER}.is_push_configured", return_value=False) + @patch(f"{CONSUMER}.get_int", return_value=5) + @patch(f"{CONSUMER}.get_bool", return_value=True) + async def test_skips_when_push_not_configured( + self, + _get_bool, + _get_int, + _configured, + mock_fetch, + mock_send, + mock_delete, + ): + join_request_id = uuid4() + device = JoinRequestPushDeviceTarget(id=uuid4(), token="tok", platform="android") + mock_fetch.return_value = _targets(join_request_id=join_request_id, devices=[device]) + + await process_join_request_notification_message( + _sqs_message(join_request_id=join_request_id) + ) + + mock_send.assert_not_called() + mock_delete.assert_called_once_with("r1") + + +class TestIdempotencyKey: + @patch(f"{CONSUMER}.get", return_value="prefix:") + def test_key_separates_event_types(self, _get): + join_request_id = uuid4() + push_device_id = uuid4() + created = _idempotency_key( + join_request_id=join_request_id, + event_type="JOIN_REQUEST_CREATED", + push_device_id=push_device_id, + ) + decided = _idempotency_key( + join_request_id=join_request_id, + event_type="JOIN_REQUEST_DECIDED", + push_device_id=push_device_id, + ) + assert created != decided + + @patch(f"{CONSUMER}.get", return_value="prefix:") + def test_key_separates_devices(self, _get): + join_request_id = uuid4() + first = _idempotency_key( + join_request_id=join_request_id, + event_type="JOIN_REQUEST_CREATED", + push_device_id=uuid4(), + ) + second = _idempotency_key( + join_request_id=join_request_id, + event_type="JOIN_REQUEST_CREATED", + push_device_id=uuid4(), + ) + assert first != second diff --git a/tests/notifications/test_join_request_sqs_client.py b/tests/notifications/test_join_request_sqs_client.py new file mode 100644 index 0000000..91a8c09 --- /dev/null +++ b/tests/notifications/test_join_request_sqs_client.py @@ -0,0 +1,179 @@ +"""Tests for join request 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.join_request_sqs_client import ( + JOIN_REQUEST_CREATED_EVENT, + JOIN_REQUEST_DECIDED_EVENT, + JOIN_REQUEST_NOTIFICATION_EVENT_VERSION, + delete_join_request_notification_message, + get_join_request_notification_sqs_queue_url, + is_join_request_notification_sqs_configured, + is_join_request_notification_sqs_poll_enabled, + parse_join_request_notification_message_body, + receive_join_request_notification_messages, +) + + +class TestQueueConfig: + @patch( + "worker_api.notifications.join_request_sqs_client.get", + return_value=" https://sqs.example/join-request ", + ) + def test_get_queue_url_strips(self, _get): + assert get_join_request_notification_sqs_queue_url() == "https://sqs.example/join-request" + + @patch( + "worker_api.notifications.join_request_sqs_client.get_join_request_notification_sqs_queue_url", + return_value="", + ) + def test_is_configured_false(self, _url): + assert is_join_request_notification_sqs_configured() is False + + @patch( + "worker_api.notifications.join_request_sqs_client.is_join_request_notification_sqs_configured", + return_value=True, + ) + @patch("worker_api.notifications.join_request_sqs_client.get_bool", return_value=True) + def test_poll_enabled(self, _bool, _configured): + assert is_join_request_notification_sqs_poll_enabled() is True + + @patch( + "worker_api.notifications.join_request_sqs_client.is_join_request_notification_sqs_configured", + return_value=False, + ) + @patch("worker_api.notifications.join_request_sqs_client.get_bool", return_value=True) + def test_poll_disabled_without_queue(self, _bool, _configured): + assert is_join_request_notification_sqs_poll_enabled() is False + + +class TestParseBody: + def test_accepts_created_event(self): + join_request_id = str(uuid4()) + body = parse_join_request_notification_message_body( + json.dumps( + { + "event_type": JOIN_REQUEST_CREATED_EVENT, + "version": JOIN_REQUEST_NOTIFICATION_EVENT_VERSION, + "join_request_id": join_request_id, + } + ) + ) + assert body["join_request_id"] == join_request_id + assert body["event_type"] == JOIN_REQUEST_CREATED_EVENT + + def test_accepts_decided_event(self): + join_request_id = str(uuid4()) + body = parse_join_request_notification_message_body( + json.dumps( + { + "event_type": JOIN_REQUEST_DECIDED_EVENT, + "version": JOIN_REQUEST_NOTIFICATION_EVENT_VERSION, + "join_request_id": join_request_id, + } + ) + ) + assert body["join_request_id"] == join_request_id + assert body["event_type"] == JOIN_REQUEST_DECIDED_EVENT + + def test_rejects_invalid_json(self): + assert parse_join_request_notification_message_body("not-json") is None + + def test_rejects_non_object_body(self): + assert parse_join_request_notification_message_body(json.dumps([1, 2, 3])) is None + + def test_rejects_unknown_event_type(self): + assert parse_join_request_notification_message_body( + json.dumps( + { + "event_type": "CHAT_MESSAGE_CREATED", + "version": 1, + "join_request_id": str(uuid4()), + } + ) + ) is None + + def test_rejects_wrong_version(self): + assert parse_join_request_notification_message_body( + json.dumps( + { + "event_type": JOIN_REQUEST_CREATED_EVENT, + "version": 99, + "join_request_id": str(uuid4()), + } + ) + ) is None + + def test_rejects_missing_join_request_id(self): + assert parse_join_request_notification_message_body( + json.dumps( + { + "event_type": JOIN_REQUEST_DECIDED_EVENT, + "version": 1, + } + ) + ) is None + + +class TestReceiveAndDelete: + @patch( + "worker_api.notifications.join_request_sqs_client.get_join_request_notification_sqs_queue_url", + return_value="", + ) + def test_receive_empty_without_queue(self, _url): + assert receive_join_request_notification_messages() == [] + + @patch("worker_api.notifications.join_request_sqs_client.get_int", side_effect=lambda key: 1) + @patch("worker_api.notifications.join_request_sqs_client._get_sqs_client") + @patch( + "worker_api.notifications.join_request_sqs_client.get_join_request_notification_sqs_queue_url", + return_value="https://sqs.example/join-request", + ) + 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_join_request_notification_messages() == [{"MessageId": "1"}] + + @patch("worker_api.notifications.join_request_sqs_client.get_int", side_effect=lambda key: 1) + @patch("worker_api.notifications.join_request_sqs_client._get_sqs_client") + @patch( + "worker_api.notifications.join_request_sqs_client.get_join_request_notification_sqs_queue_url", + return_value="https://sqs.example/join-request", + ) + def test_receive_handles_client_error(self, _url, mock_get_client, _get_int): + client = MagicMock() + client.receive_message.side_effect = ClientError( + {"Error": {"Code": "500", "Message": "fail"}}, + "ReceiveMessage", + ) + mock_get_client.return_value = client + assert receive_join_request_notification_messages() == [] + + @patch("worker_api.notifications.join_request_sqs_client._get_sqs_client") + @patch( + "worker_api.notifications.join_request_sqs_client.get_join_request_notification_sqs_queue_url", + return_value="https://sqs.example/join-request", + ) + def test_delete_message(self, _url, mock_get_client): + client = MagicMock() + mock_get_client.return_value = client + delete_join_request_notification_message("receipt") + client.delete_message.assert_called_once() + + @patch("worker_api.notifications.join_request_sqs_client._get_sqs_client") + @patch( + "worker_api.notifications.join_request_sqs_client.get_join_request_notification_sqs_queue_url", + return_value="https://sqs.example/join-request", + ) + 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_join_request_notification_message("receipt") diff --git a/worker_api/config.py b/worker_api/config.py index a6c3387..74c6456 100644 --- a/worker_api/config.py +++ b/worker_api/config.py @@ -47,6 +47,17 @@ CHAT_NOTIFICATION_IDEMPOTENCY_TTL_SECONDS=86400, CHAT_NOTIFICATION_IDEMPOTENCY_KEY_PREFIX="worker:chat-notifications:sent:", + # Join request notification SQS consumer (backend producer → worker consumer) + JOIN_REQUEST_NOTIFICATION_SQS_QUEUE_URL="", + JOIN_REQUEST_NOTIFICATION_SQS_WAIT_TIME_SECONDS=20, + JOIN_REQUEST_NOTIFICATION_SQS_VISIBILITY_TIMEOUT_SECONDS=300, + JOIN_REQUEST_NOTIFICATION_SQS_MAX_MESSAGES=5, + JOIN_REQUEST_NOTIFICATION_SQS_POLL_ENABLED="true", + JOIN_REQUEST_NOTIFICATION_SEND_CONCURRENCY=10, + JOIN_REQUEST_NOTIFICATION_TARGET_PAGE_SIZE=100, + JOIN_REQUEST_NOTIFICATION_IDEMPOTENCY_TTL_SECONDS=86400, + JOIN_REQUEST_NOTIFICATION_IDEMPOTENCY_KEY_PREFIX="worker:join-request-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 0fb9af4..6710487 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.join_request_notification_consumer import ( + run_join_request_notification_sqs_consumer, +) mongodb_client = None mongodb = None @@ -38,6 +41,7 @@ async def lifespan(api: FastAPI): consumer_tasks = [ 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)), ] yield diff --git a/worker_api/notifications/enums.py b/worker_api/notifications/enums.py index 4c3f9b9..9557d82 100644 --- a/worker_api/notifications/enums.py +++ b/worker_api/notifications/enums.py @@ -20,3 +20,4 @@ class SessionType(StrEnum): ACCUMULATION = "ACCUMULATION" TIMER = "TIMER" CHAT = "CHAT" + GROUP = "GROUP" diff --git a/worker_api/notifications/join_request_sqs_client.py b/worker_api/notifications/join_request_sqs_client.py new file mode 100644 index 0000000..40eab60 --- /dev/null +++ b/worker_api/notifications/join_request_sqs_client.py @@ -0,0 +1,107 @@ +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 + +JOIN_REQUEST_CREATED_EVENT = "JOIN_REQUEST_CREATED" +JOIN_REQUEST_DECIDED_EVENT = "JOIN_REQUEST_DECIDED" +JOIN_REQUEST_NOTIFICATION_EVENTS = frozenset( + {JOIN_REQUEST_CREATED_EVENT, JOIN_REQUEST_DECIDED_EVENT} +) +JOIN_REQUEST_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_join_request_notification_sqs_queue_url() -> str: + return get("JOIN_REQUEST_NOTIFICATION_SQS_QUEUE_URL").strip() + + +def is_join_request_notification_sqs_configured() -> bool: + return bool(get_join_request_notification_sqs_queue_url()) + + +def is_join_request_notification_sqs_poll_enabled() -> bool: + return ( + get_bool("JOIN_REQUEST_NOTIFICATION_SQS_POLL_ENABLED") + and is_join_request_notification_sqs_configured() + ) + + +def receive_join_request_notification_messages() -> List[Dict[str, Any]]: + queue_url = get_join_request_notification_sqs_queue_url() + if not queue_url: + return [] + + try: + response = _get_sqs_client().receive_message( + QueueUrl=queue_url, + MaxNumberOfMessages=get_int("JOIN_REQUEST_NOTIFICATION_SQS_MAX_MESSAGES"), + WaitTimeSeconds=get_int("JOIN_REQUEST_NOTIFICATION_SQS_WAIT_TIME_SECONDS"), + VisibilityTimeout=get_int("JOIN_REQUEST_NOTIFICATION_SQS_VISIBILITY_TIMEOUT_SECONDS"), + MessageAttributeNames=["All"], + ) + return response.get("Messages", []) + except ClientError as e: + logger.error("Failed to receive join request notification SQS messages: %s", e) + return [] + + +def delete_join_request_notification_message(receipt_handle: str) -> None: + queue_url = get_join_request_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 join request notification SQS message: %s", e) + + +def parse_join_request_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 join request notification SQS message body: %s", e) + return None + + if not isinstance(body, dict): + logger.error("Join request notification SQS message body is not an object: %s", body) + return None + + event_type = body.get("event_type") + version = body.get("version") + join_request_id = body.get("join_request_id") + if event_type not in JOIN_REQUEST_NOTIFICATION_EVENTS: + logger.error("Unsupported join request notification event_type: %s", event_type) + return None + if version != JOIN_REQUEST_NOTIFICATION_EVENT_VERSION: + logger.error("Unsupported join request notification event version: %s", version) + return None + if not join_request_id: + logger.error( + "Join request notification SQS message missing join_request_id: %s", body + ) + return None + return body diff --git a/worker_api/notifications/schemas.py b/worker_api/notifications/schemas.py index dc01b94..9c60d41 100644 --- a/worker_api/notifications/schemas.py +++ b/worker_api/notifications/schemas.py @@ -193,6 +193,33 @@ class ChatNotificationTargetsResponse(BaseModel): has_more: bool +class JoinRequestPushDeviceTarget(BaseModel): + id: UUID + token: str + platform: str + + +class JoinRequestNotificationRecipient(BaseModel): + user_id: UUID + push_devices: list[JoinRequestPushDeviceTarget] + + +class JoinRequestNotificationTargetsResponse(BaseModel): + join_request_id: UUID + group_id: UUID + event_type: str + status: str + group_name: str + requester_name: str + title: str + body: str + recipients: list[JoinRequestNotificationRecipient] + skip: int + limit: int + total: int + has_more: bool + + class DeactivatePushDeviceRequest(BaseModel): push_device_id: UUID diff --git a/worker_api/notifications/services/backend_client.py b/worker_api/notifications/services/backend_client.py index 286d519..2fbee04 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, + JoinRequestNotificationTargetsResponse, NotificationContent, RoutineNotificationTargetsResponse, VerseOfDayNotificationTargetsResponse, @@ -67,6 +68,22 @@ async def fetch_chat_notification_targets( return ChatNotificationTargetsResponse.model_validate(response.json()) +async def fetch_join_request_notification_targets( + *, + join_request_id: UUID, + skip: int = 0, + limit: int = 100, +) -> JoinRequestNotificationTargetsResponse: + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.get( + f"{_backend_url()}/internal/join-request-notification-targets/{join_request_id}", + params={"skip": skip, "limit": limit}, + headers=_backend_headers(), + ) + response.raise_for_status() + return JoinRequestNotificationTargetsResponse.model_validate(response.json()) + + async def fetch_verse_of_day_notification_targets() -> VerseOfDayNotificationTargetsResponse: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get( diff --git a/worker_api/notifications/services/join_request_notification_consumer.py b/worker_api/notifications/services/join_request_notification_consumer.py new file mode 100644 index 0000000..a8aa17f --- /dev/null +++ b/worker_api/notifications/services/join_request_notification_consumer.py @@ -0,0 +1,295 @@ +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.join_request_sqs_client import ( + delete_join_request_notification_message, + is_join_request_notification_sqs_poll_enabled, + parse_join_request_notification_message_body, + receive_join_request_notification_messages, +) +from worker_api.notifications.schemas import ( + JoinRequestNotificationTargetsResponse, + JoinRequestPushDeviceTarget, +) +from worker_api.notifications.services.backend_client import ( + deactivate_push_device, + fetch_join_request_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_join_request_push_notification, +) + +logger = logging.getLogger(__name__) + +_POLL_IDLE_SECONDS = 5 +_POLL_ERROR_SECONDS = 10 +_redis_client: redis.Redis | None = None + + +class TransientJoinRequestNotificationError(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(*, join_request_id: UUID, event_type: str, push_device_id: UUID) -> str: + prefix = get("JOIN_REQUEST_NOTIFICATION_IDEMPOTENCY_KEY_PREFIX") + return f"{prefix}{join_request_id}:{event_type}:{push_device_id}" + + +def _already_sent(*, join_request_id: UUID, event_type: str, push_device_id: UUID) -> bool: + client = _get_redis_client() + return bool( + client.exists( + _idempotency_key( + join_request_id=join_request_id, + event_type=event_type, + push_device_id=push_device_id, + ) + ) + ) + + +def _mark_sent(*, join_request_id: UUID, event_type: str, push_device_id: UUID) -> None: + client = _get_redis_client() + client.setex( + _idempotency_key( + join_request_id=join_request_id, + event_type=event_type, + push_device_id=push_device_id, + ), + get_int("JOIN_REQUEST_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(join_request_id: UUID) -> JoinRequestNotificationTargetsResponse: + page_size = max(get_int("JOIN_REQUEST_NOTIFICATION_TARGET_PAGE_SIZE"), 1) + skip = 0 + first_page: JoinRequestNotificationTargetsResponse | None = None + all_recipients = [] + + while True: + try: + page = await fetch_join_request_notification_targets( + join_request_id=join_request_id, + skip=skip, + limit=page_size, + ) + except httpx.HTTPStatusError as exc: + if exc.response.status_code == 404: + raise HTTPException(status_code=404, detail="Join request not found") from exc + raise TransientJoinRequestNotificationError(str(exc)) from exc + except httpx.HTTPError as exc: + raise TransientJoinRequestNotificationError(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: JoinRequestNotificationTargetsResponse, + device: JoinRequestPushDeviceTarget, + 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( + join_request_id=targets.join_request_id, + event_type=targets.event_type, + push_device_id=device.id, + ): + return "skipped" + + try: + await send_join_request_push_notification( + device_token=device.token, + event_type=targets.event_type, + join_request_id=targets.join_request_id, + group_id=targets.group_id, + status=targets.status, + title=targets.title, + body=targets.body, + ) + _mark_sent( + join_request_id=targets.join_request_id, + event_type=targets.event_type, + push_device_id=device.id, + ) + return "sent" + except PermanentPushTokenError: + logger.warning( + "Deactivating permanently invalid push device %s for join request %s", + device.id, + targets.join_request_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( + join_request_id=targets.join_request_id, + event_type=targets.event_type, + push_device_id=device.id, + ) + return "permanent_failed" + except Exception: + logger.exception( + "Transient FCM failure for device %s on join request %s", + device.id, + targets.join_request_id, + ) + return "transient_failed" + + +async def process_join_request_notification_message(message: Dict[str, Any]) -> None: + receipt_handle = message.get("ReceiptHandle") + body = parse_join_request_notification_message_body(message.get("Body", "")) + if not body: + if receipt_handle: + delete_join_request_notification_message(receipt_handle) + return + + join_request_id = _parse_uuid(body.get("join_request_id")) + if not join_request_id: + logger.error("Invalid join_request_id in join request notification SQS message: %s", body) + if receipt_handle: + delete_join_request_notification_message(receipt_handle) + return + + event_type = body.get("event_type") + + if not get_bool("NOTIFICATION_DISPATCH_ENABLED"): + logger.info( + "Join request notification dispatch disabled; deleting %s event for %s", + event_type, + join_request_id, + ) + if receipt_handle: + delete_join_request_notification_message(receipt_handle) + return + + try: + targets = await _fetch_all_targets(join_request_id) + except HTTPException as exc: + if exc.status_code == 404: + logger.error( + "Join request not found for notification event %s", join_request_id + ) + if receipt_handle: + delete_join_request_notification_message(receipt_handle) + return + raise TransientJoinRequestNotificationError(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 join request %s (%s); deleting event", + join_request_id, + event_type, + ) + if receipt_handle: + delete_join_request_notification_message(receipt_handle) + return + + concurrency = max(get_int("JOIN_REQUEST_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( + "Join request notification %s (%s) processed: sent=%s permanent_failed=%s transient_failed=%s skipped=%s", + join_request_id, + targets.event_type, + sent, + permanent_failed, + transient_failed, + skipped, + ) + + if transient_failed > 0: + raise TransientJoinRequestNotificationError( + f"Transient failures remain for join request {join_request_id}" + ) + + if receipt_handle: + delete_join_request_notification_message(receipt_handle) + + +async def run_join_request_notification_sqs_consumer(stop_event: asyncio.Event) -> None: + logger.info("Join request notification SQS consumer started") + while not stop_event.is_set(): + if not is_join_request_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_join_request_notification_messages) + if not messages: + continue + for message in messages: + if stop_event.is_set(): + break + try: + await process_join_request_notification_message(message) + except TransientJoinRequestNotificationError: + logger.warning( + "Leaving join request notification SQS message for retry: %s", + message.get("MessageId"), + ) + except Exception: + logger.exception("Unexpected join request notification consumer error") + except Exception: + logger.exception("Join request notification SQS consumer loop error") + try: + await asyncio.wait_for(stop_event.wait(), timeout=_POLL_ERROR_SECONDS) + except asyncio.TimeoutError: + pass + + logger.info("Join request 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 45b74fb..6044579 100644 --- a/worker_api/notifications/services/push/fcm_client.py +++ b/worker_api/notifications/services/push/fcm_client.py @@ -146,6 +146,54 @@ async def send_chat_push_notification( ) +def build_join_request_notification_data( + *, + event_type: str, + join_request_id: UUID, + group_id: UUID, + status: str, + title: str, + body: str, +) -> dict[str, str]: + """FCM data payloads require string values.""" + return { + "notification_type": event_type, + "session_type": "GROUP", + "join_request_id": str(join_request_id), + "group_id": str(group_id), + "status": status, + "source_id": str(group_id), + "title": title, + "body": body, + "image_url": "", + } + + +async def send_join_request_push_notification( + *, + device_token: str, + event_type: str, + join_request_id: UUID, + group_id: UUID, + status: str, + title: str, + body: str, +) -> None: + await send_fcm_notification( + device_token=device_token, + title=title, + body=body, + data=build_join_request_notification_data( + event_type=event_type, + join_request_id=join_request_id, + group_id=group_id, + status=status, + title=title, + body=body, + ), + ) + + def _is_permanent_token_error(exc: Exception) -> bool: if isinstance(exc, UnregisteredError): return True