diff --git a/src/govoplan_notifications/backend/manifest.py b/src/govoplan_notifications/backend/manifest.py index 99404dd..426be6c 100644 --- a/src/govoplan_notifications/backend/manifest.py +++ b/src/govoplan_notifications/backend/manifest.py @@ -34,10 +34,10 @@ def _permission(scope: str, label: str, description: str) -> PermissionDefinitio PERMISSIONS = ( - _permission(READ_SCOPE, "View notifications", "Read notification inbox, outbox, and delivery state."), + _permission(READ_SCOPE, "View notifications", "Read the authenticated actor's notification inbox and delivery state."), _permission(WRITE_SCOPE, "Manage notifications", "Create, update, cancel, read, and acknowledge notifications."), _permission(DISPATCH_SCOPE, "Dispatch notifications", "Run notification delivery attempts and delivery workers."), - _permission(ADMIN_SCOPE, "Administer notifications", "Manage tenant-wide notification delivery state."), + _permission(ADMIN_SCOPE, "Administer notifications", "Explicitly read and manage tenant-wide notification delivery state."), ) ROLE_TEMPLATES = ( diff --git a/src/govoplan_notifications/backend/router.py b/src/govoplan_notifications/backend/router.py index ecd98fe..487033c 100644 --- a/src/govoplan_notifications/backend/router.py +++ b/src/govoplan_notifications/backend/router.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import Literal + from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy.orm import Session @@ -51,17 +53,46 @@ def _response(notification) -> NotificationResponse: return NotificationResponse.model_validate(notification_response(notification)) +def _principal_recipient_ids(principal: ApiPrincipal) -> tuple[str, ...]: + """Return the stable actor identifiers accepted for personal notifications. + + Existing producers use both tenant membership/user ids and account ids. Both + identify the same authenticated actor; identity/service-account ids are + included when present so producers can address those stable identities too. + """ + + candidates = ( + getattr(principal.user, "id", None), + principal.membership_id, + principal.account_id, + principal.identity_id, + principal.principal.service_account_id, + ) + return tuple(dict.fromkeys(str(value) for value in candidates if value)) + + +def _recipient_ids_for_view(principal: ApiPrincipal, view: Literal["personal", "tenant"]) -> tuple[str, ...] | None: + if view == "tenant": + _require_scope(principal, ADMIN_SCOPE) + return None + return _principal_recipient_ids(principal) + + @router.get("", response_model=NotificationListResponse) def api_list_notifications( status_filter: str | None = Query(default=None, alias="status"), channel: str | None = None, source_module: str | None = None, recipient_id: str | None = None, + view: Literal["personal", "tenant"] = Query(default="personal"), limit: int = Query(default=100, ge=1, le=500), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> NotificationListResponse: _require_scope(principal, READ_SCOPE) + recipient_ids = _recipient_ids_for_view(principal, view) + if recipient_id is not None and recipient_ids is not None and recipient_id not in recipient_ids: + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot read another recipient's notifications") notifications = list_notifications( session, tenant_id=principal.tenant_id, @@ -69,6 +100,7 @@ def api_list_notifications( channel=channel, source_module=source_module, recipient_id=recipient_id, + recipient_ids=recipient_ids, limit=limit, ) return NotificationListResponse(notifications=[_response(notification) for notification in notifications]) @@ -76,11 +108,19 @@ def api_list_notifications( @router.get("/summary", response_model=NotificationSummaryResponse) def api_notification_summary( + view: Literal["personal", "tenant"] = Query(default="personal"), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> NotificationSummaryResponse: _require_scope(principal, READ_SCOPE) - return NotificationSummaryResponse.model_validate(notification_summary(session, tenant_id=principal.tenant_id, user_id=principal.user.id)) + return NotificationSummaryResponse.model_validate( + notification_summary( + session, + tenant_id=principal.tenant_id, + user_id=principal.user.id, + recipient_ids=_recipient_ids_for_view(principal, view), + ) + ) @router.get("/preferences/me", response_model=NotificationPreferencesResponse) @@ -137,12 +177,20 @@ def api_deliver_pending( @router.get("/{notification_id}", response_model=NotificationResponse) def api_get_notification( notification_id: str, + view: Literal["personal", "tenant"] = Query(default="personal"), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> NotificationResponse: _require_scope(principal, READ_SCOPE) try: - return _response(get_notification(session, tenant_id=principal.tenant_id, notification_id=notification_id)) + return _response( + get_notification( + session, + tenant_id=principal.tenant_id, + notification_id=notification_id, + recipient_ids=_recipient_ids_for_view(principal, view), + ) + ) except NotificationError as exc: raise _notification_http_error(exc) from exc @@ -151,12 +199,19 @@ def api_get_notification( def api_update_notification( notification_id: str, payload: NotificationUpdateRequest, + view: Literal["personal", "tenant"] = Query(default="personal"), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> NotificationResponse: _require_scope(principal, WRITE_SCOPE) try: - notification = update_notification(session, tenant_id=principal.tenant_id, notification_id=notification_id, payload=payload) + notification = update_notification( + session, + tenant_id=principal.tenant_id, + notification_id=notification_id, + payload=payload, + recipient_ids=_recipient_ids_for_view(principal, view), + ) except NotificationError as exc: raise _notification_http_error(exc) from exc session.commit() @@ -171,6 +226,7 @@ def api_deliver_notification( ) -> NotificationDeliveryResultResponse: _require_scope(principal, DISPATCH_SCOPE) try: + get_notification(session, tenant_id=principal.tenant_id, notification_id=notification_id) notification = deliver_notification(session, notification_id=notification_id) except NotificationError as exc: raise _notification_http_error(exc) from exc diff --git a/src/govoplan_notifications/backend/service.py b/src/govoplan_notifications/backend/service.py index 88f1e33..fd663b8 100644 --- a/src/govoplan_notifications/backend/service.py +++ b/src/govoplan_notifications/backend/service.py @@ -6,7 +6,7 @@ from email.message import EmailMessage from pathlib import Path from typing import Any -from sqlalchemy import or_ +from sqlalchemy import event, or_ from sqlalchemy.orm import Session from govoplan_core.core.notifications import NotificationDispatchRequest @@ -20,6 +20,34 @@ class NotificationError(ValueError): PENDING_STATUSES = {"pending", "queued", "failed"} +_AFTER_COMMIT_DELIVERY_IDS = "govoplan_notifications_after_commit_delivery_ids" + + +@event.listens_for(Session, "after_commit") +def _enqueue_committed_notifications(session: Session) -> None: + notification_ids = tuple(session.info.pop(_AFTER_COMMIT_DELIVERY_IDS, ())) + for notification_id in notification_ids: + _enqueue_celery_delivery(notification_id) + + +@event.listens_for(Session, "after_rollback") +def _discard_rolled_back_notifications(session: Session) -> None: + session.info.pop(_AFTER_COMMIT_DELIVERY_IDS, None) + + +def _enqueue_notification_after_commit(session: Session, notification_id: str) -> None: + pending = session.info.setdefault(_AFTER_COMMIT_DELIVERY_IDS, []) + if notification_id not in pending: + pending.append(notification_id) + + +def _discard_notification_after_commit(session: Session, notification_id: str) -> None: + pending = session.info.get(_AFTER_COMMIT_DELIVERY_IDS) + if not isinstance(pending, list) or notification_id not in pending: + return + pending.remove(notification_id) + if not pending: + session.info.pop(_AFTER_COMMIT_DELIVERY_IDS, None) def response_datetime(value: datetime | None) -> datetime | None: @@ -87,7 +115,7 @@ def create_notification( session.add(notification) session.flush() if payload.enqueue_delivery and notification.status == "queued": - _enqueue_celery_delivery(notification.id) + _enqueue_notification_after_commit(session, notification.id) return notification @@ -131,6 +159,7 @@ def list_notifications( channel: str | None = None, source_module: str | None = None, recipient_id: str | None = None, + recipient_ids: tuple[str, ...] | None = None, limit: int = 100, ) -> list[NotificationMessage]: query = session.query(NotificationMessage).filter( @@ -143,16 +172,26 @@ def list_notifications( query = query.filter(NotificationMessage.channel == _clean_channel(channel)) if source_module: query = query.filter(NotificationMessage.source_module == source_module) + if recipient_ids is not None: + query = query.filter(NotificationMessage.recipient_id.in_(recipient_ids)) if recipient_id: query = query.filter(NotificationMessage.recipient_id == recipient_id) return query.order_by(NotificationMessage.created_at.desc(), NotificationMessage.id.asc()).limit(limit).all() -def notification_summary(session: Session, *, tenant_id: str, user_id: str | None = None) -> dict[str, int | bool]: +def notification_summary( + session: Session, + *, + tenant_id: str, + user_id: str | None = None, + recipient_ids: tuple[str, ...] | None = None, +) -> dict[str, int | bool]: base_query = session.query(NotificationMessage).filter( NotificationMessage.tenant_id == tenant_id, NotificationMessage.deleted_at.is_(None), ) + if recipient_ids is not None: + base_query = base_query.filter(NotificationMessage.recipient_id.in_(recipient_ids)) active_query = base_query.filter(NotificationMessage.status.notin_(["cancelled", "skipped"])) show_unread_badge = True if user_id: @@ -166,15 +205,22 @@ def notification_summary(session: Session, *, tenant_id: str, user_id: str | Non } -def get_notification(session: Session, *, tenant_id: str, notification_id: str) -> NotificationMessage: +def get_notification( + session: Session, + *, + tenant_id: str, + notification_id: str, + recipient_ids: tuple[str, ...] | None = None, +) -> NotificationMessage: + query = session.query(NotificationMessage).filter( + NotificationMessage.tenant_id == tenant_id, + NotificationMessage.id == notification_id, + NotificationMessage.deleted_at.is_(None), + ) + if recipient_ids is not None: + query = query.filter(NotificationMessage.recipient_id.in_(recipient_ids)) notification = ( - session.query(NotificationMessage) - .filter( - NotificationMessage.tenant_id == tenant_id, - NotificationMessage.id == notification_id, - NotificationMessage.deleted_at.is_(None), - ) - .first() + query.first() ) if notification is None: raise NotificationError("Notification not found") @@ -187,8 +233,14 @@ def update_notification( tenant_id: str, notification_id: str, payload: NotificationUpdateRequest, + recipient_ids: tuple[str, ...] | None = None, ) -> NotificationMessage: - notification = get_notification(session, tenant_id=tenant_id, notification_id=notification_id) + notification = get_notification( + session, + tenant_id=tenant_id, + notification_id=notification_id, + recipient_ids=recipient_ids, + ) now = _now() if payload.status == "read": notification.read_at = notification.read_at or now @@ -210,6 +262,9 @@ def deliver_notification( notification_id: str, settings: object | None = None, ) -> NotificationMessage: + # An explicit same-transaction delivery supersedes the deferred Celery + # handoff registered when the row was created. + _discard_notification_after_commit(session, notification_id) notification = session.get(NotificationMessage, notification_id) if notification is None or notification.deleted_at is not None: raise NotificationError("Notification not found") @@ -404,11 +459,14 @@ def _enqueue_celery_delivery(notification_id: str) -> None: try: from govoplan_core.celery_app import celery from govoplan_core.settings import settings + if not getattr(settings, "celery_enabled", False): + return + celery.send_task("govoplan.notifications.deliver", args=[notification_id], queue="notifications") except Exception: + # The committed queued row is the durable source of truth. A broker or + # import failure must not turn a successful database commit into an API + # failure; operators/workers can recover it through deliver_pending. return - if not getattr(settings, "celery_enabled", False): - return - celery.send_task("govoplan.notifications.deliver", args=[notification_id], queue="notifications") def notification_response(notification: NotificationMessage) -> dict[str, Any]: diff --git a/tests/test_notifications.py b/tests/test_notifications.py index 3c56b60..0e954dd 100644 --- a/tests/test_notifications.py +++ b/tests/test_notifications.py @@ -2,16 +2,22 @@ from __future__ import annotations import tempfile import unittest +from unittest.mock import patch +from types import SimpleNamespace +from fastapi import HTTPException from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef from govoplan_core.core.modules import ModuleContext from govoplan_core.core.notifications import NotificationDispatchRequest from govoplan_core.db.base import Base from govoplan_notifications.backend.capabilities import dispatch_capability from govoplan_notifications.backend.db.models import NotificationDeliveryAttempt, NotificationMessage, NotificationPreference -from govoplan_notifications.backend.schemas import NotificationCreateRequest, NotificationPreferencesUpdateRequest +from govoplan_notifications.backend.router import api_get_notification, api_list_notifications, api_notification_summary, api_update_notification +from govoplan_notifications.backend.schemas import NotificationCreateRequest, NotificationPreferencesUpdateRequest, NotificationUpdateRequest from govoplan_notifications.backend.service import create_notification, deliver_pending, notification_preferences_response, notification_summary, update_notification_preferences @@ -21,25 +27,212 @@ class NotificationServiceTests(unittest.TestCase): Base.metadata.create_all(self.engine, tables=[NotificationMessage.__table__, NotificationDeliveryAttempt.__table__, NotificationPreference.__table__]) self.Session = sessionmaker(bind=self.engine) - def test_create_and_deliver_inbox_notification(self) -> None: + def tearDown(self) -> None: + Base.metadata.drop_all( + self.engine, + tables=[NotificationDeliveryAttempt.__table__, NotificationMessage.__table__, NotificationPreference.__table__], + ) + self.engine.dispose() + + @staticmethod + def _principal(*, tenant_id: str, user_id: str, account_id: str, scopes: set[str]) -> ApiPrincipal: + user = SimpleNamespace(id=user_id) + return ApiPrincipal( + principal=PrincipalRef( + account_id=account_id, + membership_id=user_id, + tenant_id=tenant_id, + scopes=frozenset(scopes), + ), + account=SimpleNamespace(id=account_id), + user=user, + ) + + @staticmethod + def _inbox_payload(*, recipient_id: str, subject: str) -> NotificationCreateRequest: + return NotificationCreateRequest( + source_module="test", + source_resource_type="thing", + event_kind="created", + channel="inbox", + recipient_type="user", + recipient_id=recipient_id, + subject=subject, + enqueue_delivery=False, + ) + + def test_personal_inbox_cannot_read_or_update_another_recipient(self) -> None: with self.Session() as session: - notification = create_notification( + own_membership = create_notification( session, tenant_id="tenant-1", - payload=NotificationCreateRequest( - source_module="test", - source_resource_type="thing", - source_resource_id="thing-1", - event_kind="created", - channel="inbox", - recipient_id="user-1", - subject="Created", - ), + payload=self._inbox_payload(recipient_id="user-1", subject="Membership addressed"), ) - result = deliver_pending(session, tenant_id="tenant-1") - self.assertEqual(result["sent"], 1) - self.assertEqual(notification.status, "sent") - self.assertEqual(notification.attempt_count, 1) + own_account = create_notification( + session, + tenant_id="tenant-1", + payload=self._inbox_payload(recipient_id="account-1", subject="Account addressed"), + ) + another_user = create_notification( + session, + tenant_id="tenant-1", + payload=self._inbox_payload(recipient_id="user-2", subject="Private for another user"), + ) + other_tenant = create_notification( + session, + tenant_id="tenant-2", + payload=self._inbox_payload(recipient_id="user-1", subject="Other tenant"), + ) + principal = self._principal( + tenant_id="tenant-1", + user_id="user-1", + account_id="account-1", + scopes={"notifications:notification:read", "notifications:notification:write"}, + ) + + result = api_list_notifications( + status_filter=None, + channel=None, + source_module=None, + recipient_id=None, + view="personal", + limit=100, + session=session, + principal=principal, + ) + self.assertEqual({own_membership.id, own_account.id}, {item.id for item in result.notifications}) + + summary = api_notification_summary(view="personal", session=session, principal=principal) + self.assertEqual(summary.total, 2) + + with self.assertRaises(HTTPException) as hidden_read: + api_get_notification(another_user.id, view="personal", session=session, principal=principal) + self.assertEqual(hidden_read.exception.status_code, 404) + + with self.assertRaises(HTTPException) as recipient_impersonation: + api_list_notifications( + status_filter=None, + channel=None, + source_module=None, + recipient_id="user-2", + view="personal", + limit=100, + session=session, + principal=principal, + ) + self.assertEqual(recipient_impersonation.exception.status_code, 403) + + with self.assertRaises(HTTPException) as hidden_update: + api_update_notification( + another_user.id, + NotificationUpdateRequest(status="read"), + view="personal", + session=session, + principal=principal, + ) + self.assertEqual(hidden_update.exception.status_code, 404) + self.assertIsNone(another_user.read_at) + + with self.assertRaises(HTTPException) as cross_tenant: + api_get_notification(other_tenant.id, view="personal", session=session, principal=principal) + self.assertEqual(cross_tenant.exception.status_code, 404) + + def test_tenant_notification_view_is_an_explicit_admin_operation(self) -> None: + with self.Session() as session: + another_user = create_notification( + session, + tenant_id="tenant-1", + payload=self._inbox_payload(recipient_id="user-2", subject="Administrative outbox entry"), + ) + reader = self._principal( + tenant_id="tenant-1", + user_id="user-1", + account_id="account-1", + scopes={"notifications:notification:read"}, + ) + admin = self._principal( + tenant_id="tenant-1", + user_id="user-admin", + account_id="account-admin", + scopes={"notifications:notification:read", "notifications:notification:admin"}, + ) + + with self.assertRaises(HTTPException) as forbidden: + api_get_notification(another_user.id, view="tenant", session=session, principal=reader) + self.assertEqual(forbidden.exception.status_code, 403) + + visible = api_get_notification(another_user.id, view="tenant", session=session, principal=admin) + self.assertEqual(visible.id, another_user.id) + + def test_create_and_deliver_inbox_notification(self) -> None: + with patch("govoplan_notifications.backend.service._enqueue_celery_delivery") as enqueue: + with self.Session() as session: + notification = create_notification( + session, + tenant_id="tenant-1", + payload=NotificationCreateRequest( + source_module="test", + source_resource_type="thing", + source_resource_id="thing-1", + event_kind="created", + channel="inbox", + recipient_id="user-1", + subject="Created", + ), + ) + result = deliver_pending(session, tenant_id="tenant-1") + self.assertEqual(result["sent"], 1) + self.assertEqual(notification.status, "sent") + self.assertEqual(notification.attempt_count, 1) + session.commit() + enqueue.assert_not_called() + + def test_delivery_task_is_enqueued_only_after_notification_commit(self) -> None: + with patch("govoplan_notifications.backend.service._enqueue_celery_delivery") as enqueue: + with self.Session() as session: + notification = create_notification( + session, + tenant_id="tenant-1", + payload=NotificationCreateRequest( + source_module="test", + source_resource_type="thing", + event_kind="created", + channel="inbox", + recipient_id="user-1", + subject="Committed", + ), + ) + notification_id = notification.id + enqueue.assert_not_called() + + session.commit() + + enqueue.assert_called_once_with(notification_id) + with self.Session() as verification_session: + self.assertIsNotNone(verification_session.get(NotificationMessage, notification_id)) + + def test_rolled_back_notification_never_enqueues_a_delivery_task(self) -> None: + with patch("govoplan_notifications.backend.service._enqueue_celery_delivery") as enqueue: + with self.Session() as session: + notification = create_notification( + session, + tenant_id="tenant-1", + payload=NotificationCreateRequest( + source_module="test", + source_resource_type="thing", + event_kind="created", + channel="inbox", + recipient_id="user-1", + subject="Rolled back", + ), + ) + notification_id = notification.id + session.rollback() + session.commit() + + enqueue.assert_not_called() + with self.Session() as verification_session: + self.assertIsNone(verification_session.get(NotificationMessage, notification_id)) def test_mail_delivery_uses_local_file_transport(self) -> None: with tempfile.TemporaryDirectory() as tmpdir, self.Session() as session: diff --git a/webui/src/api/notifications.ts b/webui/src/api/notifications.ts index b6bdc66..94a7027 100644 --- a/webui/src/api/notifications.ts +++ b/webui/src/api/notifications.ts @@ -82,12 +82,13 @@ export type NotificationDeliveryResult = { errors: string[]; }; -export function listNotifications(settings: ApiSettings, filters: { status?: string; channel?: string; source_module?: string; recipient_id?: string; limit?: number } = {}): Promise { +export function listNotifications(settings: ApiSettings, filters: { status?: string; channel?: string; source_module?: string; recipient_id?: string; view?: "personal" | "tenant"; limit?: number } = {}): Promise { const params = new URLSearchParams(); if (filters.status) params.set("status", filters.status); if (filters.channel) params.set("channel", filters.channel); if (filters.source_module) params.set("source_module", filters.source_module); if (filters.recipient_id) params.set("recipient_id", filters.recipient_id); + if (filters.view) params.set("view", filters.view); if (filters.limit) params.set("limit", String(filters.limit)); const query = params.toString(); return apiFetch(settings, `/api/v1/notifications${query ? `?${query}` : ""}`);