feat(notifications): add governed DSAR coverage

src/govoplan_notifications/backend/dsar_provider.py

@@ -0,0 +1,655 @@
+from __future__ import annotations
+
+import math
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from sqlalchemy import func, or_
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_notifications.backend.db.models import (
+    NotificationMessage,
+    NotificationPreference,
+)
+
+
+NOTIFICATIONS_DSAR_CAPABILITY = dsar_capability_name("notifications")
+_MAX_RECORDS = 1_000
+_MAX_ATTEMPTS_PER_MESSAGE = 100
+_MAX_BODY_CHARS = 100_000
+_MAX_JSON_NODES = 10_000
+_MAX_JSON_CHARS = 100_000
+_CONFLICT = object()
+_SENSITIVE_KEY_PARTS = (
+    "authorization",
+    "cookie",
+    "credential",
+    "password",
+    "secret",
+    "token",
+    "api_key",
+    "apikey",
+)
+
+
+@dataclass(frozen=True, slots=True)
+class _SubjectSelectors:
+    actor_ids: tuple[str, ...]
+    email: str | None
+    notification_id: str | None
+    preference_id: str | None
+
+
+@dataclass(slots=True)
+class _JsonBudget:
+    nodes: int = 0
+    characters: int = 0
+
+
+class NotificationsDsarProvider:
+    provider_id = "notifications"
+    module_id = "notifications"
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]:
+        db = _session(session)
+        selectors = _subject_selectors(subject)
+        if selectors is None:
+            return ()
+
+        records: list[DsarRecordRef] = []
+        if selectors.preference_id is None:
+            messages = db.query(NotificationMessage).filter(
+                NotificationMessage.tenant_id == tenant_id,
+                _recipient_filter(selectors),
+            )
+            if selectors.notification_id:
+                messages = messages.filter(
+                    NotificationMessage.id == selectors.notification_id
+                )
+            rows = (
+                messages.order_by(
+                    NotificationMessage.created_at,
+                    NotificationMessage.id,
+                )
+                .limit(_MAX_RECORDS + 1)
+                .all()
+            )
+            if len(rows) > _MAX_RECORDS:
+                raise ValueError(
+                    "Notifications DSAR result limit exceeded; narrow the selectors."
+                )
+            records.extend(_message_record(row) for row in rows)
+
+        if selectors.notification_id is None and selectors.actor_ids:
+            preferences = db.query(NotificationPreference).filter(
+                NotificationPreference.tenant_id == tenant_id,
+                NotificationPreference.user_id.in_(selectors.actor_ids),
+            )
+            if selectors.preference_id:
+                preferences = preferences.filter(
+                    NotificationPreference.id == selectors.preference_id
+                )
+            rows = (
+                preferences.order_by(NotificationPreference.id)
+                .limit(_MAX_RECORDS + 1)
+                .all()
+            )
+            if len(rows) > _MAX_RECORDS:
+                raise ValueError(
+                    "Notification preference DSAR result limit exceeded; narrow the selectors."
+                )
+            records.extend(_preference_record(row) for row in rows)
+
+        if len(records) > _MAX_RECORDS:
+            raise ValueError(
+                "Notifications DSAR combined result limit exceeded; narrow the selectors."
+            )
+        order = {"notification_preference": 10, "notification_message": 20}
+        return tuple(
+            sorted(
+                records,
+                key=lambda item: (order[item.resource_type], item.resource_id),
+            )
+        )
+
+    def plan_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        records: Sequence[DsarRecordRef],
+    ) -> Sequence[DsarErasureActionRef]:
+        db = _session(session)
+        selectors = _subject_selectors(subject)
+        if selectors is None:
+            raise ValueError("Notifications DSAR subject selectors conflict.")
+
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            _validate_record(record)
+            if record.resource_type == "notification_preference":
+                preference = (
+                    db.query(NotificationPreference)
+                    .filter(
+                        NotificationPreference.tenant_id == tenant_id,
+                        NotificationPreference.id == record.resource_id,
+                    )
+                    .one_or_none()
+                )
+                if (
+                    preference is None
+                    or preference.user_id not in selectors.actor_ids
+                    or (
+                        selectors.preference_id
+                        and preference.id != selectors.preference_id
+                    )
+                ):
+                    actions.append(
+                        _manual_action(record, "preference ownership mismatch")
+                    )
+                    continue
+                actions.append(_delete_action(record, preference.updated_at))
+                continue
+
+            notification = (
+                db.query(NotificationMessage)
+                .filter(
+                    NotificationMessage.tenant_id == tenant_id,
+                    NotificationMessage.id == record.resource_id,
+                    _recipient_filter(selectors),
+                )
+                .one_or_none()
+            )
+            if notification is None or (
+                selectors.notification_id
+                and notification.id != selectors.notification_id
+            ):
+                actions.append(_manual_action(record, "recipient or selector mismatch"))
+            elif notification.channel == "inbox":
+                actions.append(_delete_action(record, notification.updated_at))
+            else:
+                actions.append(
+                    _manual_action(
+                        record,
+                        "external-channel delivery and provider evidence require retention review",
+                    )
+                )
+        return tuple(actions)
+
+    def execute_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        actions: Sequence[DsarErasureActionRef],
+        request_id: str,
+    ) -> Sequence[DsarExecutionResultRef]:
+        db = _session(session)
+        selectors = _subject_selectors(subject)
+        if selectors is None:
+            raise ValueError("Notifications DSAR subject selectors conflict.")
+
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if not action.executable or action.kind != "delete":
+                results.append(
+                    DsarExecutionResultRef(
+                        action_id=action.action_id,
+                        status="blocked",
+                        summary=(
+                            "External notification delivery evidence requires an "
+                            "authorized retention review."
+                        ),
+                    )
+                )
+                continue
+
+            if action.resource_type == "notification_preference":
+                row = (
+                    db.query(NotificationPreference)
+                    .filter(
+                        NotificationPreference.tenant_id == tenant_id,
+                        NotificationPreference.id == action.resource_id,
+                    )
+                    .one_or_none()
+                )
+                if row is None:
+                    results.append(_unchanged(action, request_id, "preference"))
+                    continue
+                if (
+                    row.user_id not in selectors.actor_ids
+                    or str(action.metadata.get("owner_id") or "") != row.user_id
+                    or (selectors.preference_id and row.id != selectors.preference_id)
+                ):
+                    results.append(_blocked(action, "preference ownership mismatch"))
+                    continue
+            elif action.resource_type == "notification_message":
+                row = (
+                    db.query(NotificationMessage)
+                    .filter(
+                        NotificationMessage.tenant_id == tenant_id,
+                        NotificationMessage.id == action.resource_id,
+                    )
+                    .one_or_none()
+                )
+                if row is None:
+                    results.append(_unchanged(action, request_id, "notification"))
+                    continue
+                if (
+                    row.channel != "inbox"
+                    or not _recipient_matches(row, selectors)
+                    or (
+                        selectors.notification_id
+                        and row.id != selectors.notification_id
+                    )
+                ):
+                    results.append(_blocked(action, "recipient or channel mismatch"))
+                    continue
+            else:
+                raise ValueError("Unsupported executable Notifications DSAR action.")
+
+            if action.metadata.get("updated_at") != _iso(row.updated_at):
+                results.append(_blocked(action, "the record changed after planning"))
+                continue
+            resource_id = str(row.id)
+            db.delete(row)
+            db.flush()
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="executed",
+                    summary=(
+                        "Deleted the personal Notifications record without "
+                        "traversing or changing its source-module data."
+                    ),
+                    evidence={
+                        "request_id": request_id,
+                        "resource_type": action.resource_type,
+                        "resource_id": resource_id,
+                    },
+                )
+            )
+        return tuple(results)
+
+
+def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
+    references = subject.external_references
+    values = {
+        "account_id": _coalesce(
+            subject.account_id,
+            references.get("notifications.account"),
+            references.get("access.account"),
+        ),
+        "membership_id": _coalesce(
+            subject.membership_id,
+            references.get("notifications.membership"),
+            references.get("tenancy.membership"),
+        ),
+        "identity_id": _coalesce(
+            subject.identity_id,
+            references.get("notifications.identity"),
+            references.get("identity.id"),
+        ),
+        "user_id": _coalesce(
+            references.get("notifications.user"),
+            references.get("idm.user"),
+        ),
+        "email": _coalesce(
+            _normalized_email(subject.email),
+            _normalized_email(references.get("notifications.email")),
+        ),
+        "notification_id": _coalesce(
+            references.get("notifications.notification"),
+            references.get("notifications.message"),
+        ),
+        "preference_id": _coalesce(
+            references.get("notifications.preference"),
+            references.get("notifications.preference_id"),
+        ),
+    }
+    if any(value is _CONFLICT for value in values.values()):
+        return None
+    actor_ids = tuple(
+        dict.fromkeys(
+            value
+            for key in ("account_id", "membership_id", "identity_id", "user_id")
+            if (value := _optional_string(values[key]))
+        )
+    )
+    email = _optional_string(values["email"])
+    if not actor_ids and email is None:
+        return None
+    return _SubjectSelectors(
+        actor_ids=actor_ids,
+        email=email,
+        notification_id=_optional_string(values["notification_id"]),
+        preference_id=_optional_string(values["preference_id"]),
+    )
+
+
+def _coalesce(*values: str | None) -> str | None | object:
+    normalized = {str(value).strip() for value in values if str(value or "").strip()}
+    if len(normalized) > 1:
+        return _CONFLICT
+    return next(iter(normalized), None)
+
+
+def _optional_string(value: object) -> str | None:
+    return value if isinstance(value, str) and value else None
+
+
+def _normalized_email(value: str | None) -> str | None:
+    normalized = (value or "").strip().casefold()
+    return normalized or None
+
+
+def _recipient_filter(selectors: _SubjectSelectors):
+    filters = []
+    if selectors.actor_ids:
+        filters.append(NotificationMessage.recipient_id.in_(selectors.actor_ids))
+    if selectors.email:
+        filters.extend(
+            (
+                func.lower(NotificationMessage.recipient) == selectors.email,
+                func.lower(NotificationMessage.recipient_id) == selectors.email,
+            )
+        )
+    return or_(*filters)
+
+
+def _recipient_matches(
+    notification: NotificationMessage,
+    selectors: _SubjectSelectors,
+) -> bool:
+    if notification.recipient_id in selectors.actor_ids:
+        return True
+    return bool(
+        selectors.email
+        and selectors.email
+        in {
+            _normalized_email(notification.recipient),
+            _normalized_email(notification.recipient_id),
+        }
+    )
+
+
+def _message_record(notification: NotificationMessage) -> DsarRecordRef:
+    attempts = list(notification.attempts)
+    if len(attempts) > _MAX_ATTEMPTS_PER_MESSAGE:
+        raise ValueError(
+            "Notifications DSAR delivery-attempt limit exceeded; narrow the selectors."
+        )
+    return DsarRecordRef(
+        provider_id="notifications",
+        module_id="notifications",
+        resource_type="notification_message",
+        resource_id=notification.id,
+        category="direct_notification",
+        title=f"Notification: {(notification.subject or notification.event_kind)[:200]}",
+        data={
+            "source_module": notification.source_module,
+            "source_resource_type": notification.source_resource_type,
+            "source_resource_id": notification.source_resource_id,
+            "event_kind": notification.event_kind,
+            "channel": notification.channel,
+            "recipient": notification.recipient,
+            "recipient_type": notification.recipient_type,
+            "recipient_id": notification.recipient_id,
+            "recipient_label": notification.recipient_label,
+            "subject": notification.subject,
+            "body_text": _bounded_text(notification.body_text, "body text"),
+            "body_html": _bounded_text(notification.body_html, "body HTML"),
+            "action_url": notification.action_url,
+            "priority": notification.priority,
+            "status": notification.status,
+            "not_before_at": _iso(notification.not_before_at),
+            "queued_at": _iso(notification.queued_at),
+            "sent_at": _iso(notification.sent_at),
+            "failed_at": _iso(notification.failed_at),
+            "read_at": _iso(notification.read_at),
+            "acknowledged_at": _iso(notification.acknowledged_at),
+            "cancelled_at": _iso(notification.cancelled_at),
+            "attempt_count": notification.attempt_count,
+            "last_error": _bounded_text(notification.last_error, "last error"),
+            "external_message_id": notification.external_message_id,
+            "payload": _bounded_json(notification.payload, label="payload"),
+            "metadata": _bounded_json(notification.metadata_, label="metadata"),
+            "deleted_at": _iso(notification.deleted_at),
+            "attempts": [
+                {
+                    "id": attempt.id,
+                    "attempt_no": attempt.attempt_no,
+                    "channel": attempt.channel,
+                    "provider": attempt.provider,
+                    "status": attempt.status,
+                    "started_at": _iso(attempt.started_at),
+                    "finished_at": _iso(attempt.finished_at),
+                    "external_message_id": attempt.external_message_id,
+                    "error": _bounded_text(attempt.error, "attempt error"),
+                    "details": _bounded_json(
+                        attempt.details,
+                        label="attempt details",
+                    ),
+                }
+                for attempt in attempts
+            ],
+        },
+        observed_at=_aware(notification.updated_at),
+        retention_reason=(
+            "External-channel messages require delivery and source-record retention review."
+            if notification.channel != "inbox"
+            else None
+        ),
+    )
+
+
+def _preference_record(preference: NotificationPreference) -> DsarRecordRef:
+    muted = preference.muted_source_modules
+    if not isinstance(muted, list) or len(muted) > 500:
+        raise ValueError("Notification muted-source preference exceeds the DSAR bound.")
+    return DsarRecordRef(
+        provider_id="notifications",
+        module_id="notifications",
+        resource_type="notification_preference",
+        resource_id=preference.id,
+        category="personal_communication_preference",
+        title="Personal notification preferences",
+        data={
+            "user_id": preference.user_id,
+            "show_unread_badge": preference.show_unread_badge,
+            "email_enabled": preference.email_enabled,
+            "email_digest_enabled": preference.email_digest_enabled,
+            "muted_source_modules": [str(value)[:100] for value in muted],
+            "metadata": _bounded_json(
+                preference.metadata_,
+                label="preference metadata",
+            ),
+        },
+        observed_at=_aware(preference.updated_at),
+    )
+
+
+def _bounded_text(value: str | None, label: str) -> str | None:
+    if value is None:
+        return None
+    if len(value) > _MAX_BODY_CHARS:
+        raise ValueError(
+            f"Notification {label} exceeds the DSAR export bound; narrow and review the record."
+        )
+    return value
+
+
+def _bounded_json(value: object, *, label: str) -> object:
+    budget = _JsonBudget()
+
+    def project(item: object, depth: int) -> object:
+        budget.nodes += 1
+        if budget.nodes > _MAX_JSON_NODES or depth > 12:
+            raise ValueError(f"Notification {label} exceeds the DSAR structure bound.")
+        if item is None or isinstance(item, (bool, int)):
+            return item
+        if isinstance(item, float):
+            if not math.isfinite(item):
+                raise ValueError(f"Notification {label} contains a non-finite number.")
+            return item
+        if isinstance(item, str):
+            budget.characters += len(item)
+            if budget.characters > _MAX_JSON_CHARS:
+                raise ValueError(
+                    f"Notification {label} exceeds the DSAR character bound."
+                )
+            return item
+        if isinstance(item, Mapping):
+            if len(item) > 1_000:
+                raise ValueError(f"Notification {label} contains too many fields.")
+            projected: dict[str, object] = {}
+            for raw_key, raw_value in item.items():
+                key = str(raw_key)
+                if len(key) > 500:
+                    raise ValueError(f"Notification {label} contains an oversized key.")
+                budget.characters += len(key)
+                if budget.characters > _MAX_JSON_CHARS:
+                    raise ValueError(
+                        f"Notification {label} exceeds the DSAR character bound."
+                    )
+                projected[key] = (
+                    "[redacted]"
+                    if _is_sensitive_key(key)
+                    else project(raw_value, depth + 1)
+                )
+            return projected
+        if isinstance(item, (list, tuple)):
+            if len(item) > 1_000:
+                raise ValueError(f"Notification {label} contains too many items.")
+            return [project(entry, depth + 1) for entry in item]
+        raise ValueError(f"Notification {label} contains an unsupported value.")
+
+    return project(value, 0)
+
+
+def _is_sensitive_key(value: str) -> bool:
+    normalized = value.strip().casefold().replace("-", "_")
+    return any(part in normalized for part in _SENSITIVE_KEY_PARTS)
+
+
+def _delete_action(
+    record: DsarRecordRef,
+    updated_at: datetime | None,
+) -> DsarErasureActionRef:
+    metadata: dict[str, object] = {"updated_at": _iso(updated_at)}
+    if record.resource_type == "notification_preference":
+        metadata["owner_id"] = str(record.data.get("user_id") or "")
+    return DsarErasureActionRef(
+        action_id=f"notifications:delete:{record.resource_type}:{record.resource_id}",
+        provider_id="notifications",
+        module_id="notifications",
+        kind="delete",
+        resource_type=record.resource_type,
+        resource_id=record.resource_id,
+        title=f"Delete {record.title}",
+        rationale=(
+            "The record is a subject-owned preference or derived in-product "
+            "notification copy; source-module data remains unchanged."
+        ),
+        executable=True,
+        irreversible=True,
+        metadata=metadata,
+    )
+
+
+def _manual_action(record: DsarRecordRef, reason: str) -> DsarErasureActionRef:
+    return DsarErasureActionRef(
+        action_id=(
+            f"notifications:manual_review:{record.resource_type}:{record.resource_id}"
+        ),
+        provider_id="notifications",
+        module_id="notifications",
+        kind="manual_review",
+        resource_type=record.resource_type,
+        resource_id=record.resource_id,
+        title=f"Review {record.title}",
+        rationale=f"Automatic deletion stopped because {reason}.",
+        executable=False,
+    )
+
+
+def _unchanged(
+    action: DsarErasureActionRef,
+    request_id: str,
+    label: str,
+) -> DsarExecutionResultRef:
+    return DsarExecutionResultRef(
+        action_id=action.action_id,
+        status="unchanged",
+        summary=f"The personal notification {label} was already absent.",
+        evidence={"request_id": request_id},
+    )
+
+
+def _blocked(
+    action: DsarErasureActionRef,
+    reason: str,
+) -> DsarExecutionResultRef:
+    return DsarExecutionResultRef(
+        action_id=action.action_id,
+        status="blocked",
+        summary=f"Notifications erasure stopped because of {reason}.",
+    )
+
+
+def _iso(value: datetime | None) -> str | None:
+    aware = _aware(value)
+    return aware.isoformat() if aware else None
+
+
+def _aware(value: datetime | None) -> datetime | None:
+    if value is None or value.tzinfo is not None:
+        return value
+    return value.replace(tzinfo=timezone.utc)
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Notifications DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "notifications" or record.module_id != "notifications":
+        raise ValueError("Notifications DSAR cannot plan a foreign provider record.")
+    if (
+        record.resource_type
+        not in {
+            "notification_message",
+            "notification_preference",
+        }
+        or not record.resource_id
+    ):
+        raise ValueError("Notifications DSAR record identity is invalid.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "notifications" or action.module_id != "notifications":
+        raise ValueError("Notifications DSAR cannot execute a foreign provider action.")
+    if not action.action_id.startswith("notifications:"):
+        raise ValueError("Notifications DSAR action identity is invalid.")
+
+
+__all__ = ["NOTIFICATIONS_DSAR_CAPABILITY", "NotificationsDsarProvider"]

src/govoplan_notifications/backend/manifest.py

@@ -2,9 +2,16 @@ from __future__ import annotations
 
 from pathlib import Path
 
-from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
+from govoplan_core.core.access import (
-from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard
+    CAPABILITY_AUTH_PERMISSION_EVALUATOR,
+    CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
+)
+from govoplan_core.core.module_guards import (
+    drop_table_retirement_provider,
+    persistent_table_uninstall_guard,
+)
 from govoplan_core.core.modules import (
+    CapabilityDocumentation,
     DocumentationTopic,
     FrontendModule,
     FrontendRoute,
@@ -21,6 +28,10 @@ from govoplan_core.core.notifications import CAPABILITY_NOTIFICATIONS_DISPATCH
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_notifications.backend.db import models as notification_models  # noqa: F401 - populate Notifications ORM metadata
+from govoplan_notifications.backend.dsar_provider import (
+    NOTIFICATIONS_DSAR_CAPABILITY,
+    NotificationsDsarProvider,
+)
 
 
 MODULE_ID = "notifications"
@@ -47,10 +58,26 @@ def _permission(scope: str, label: str, description: str) -> PermissionDefinitio
 
 
 PERMISSIONS = (
-    _permission(READ_SCOPE, "View notifications", "Read the authenticated actor's notification inbox and delivery state."),
+    _permission(
-    _permission(WRITE_SCOPE, "Manage notifications", "Create, update, cancel, read, and acknowledge notifications."),
+        READ_SCOPE,
-    _permission(DISPATCH_SCOPE, "Dispatch notifications", "Run notification delivery attempts and delivery workers."),
+        "View notifications",
-    _permission(ADMIN_SCOPE, "Administer notifications", "Explicitly read and manage tenant-wide notification delivery state."),
+        "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",
+        "Explicitly read and manage tenant-wide notification delivery state.",
+    ),
 )
 
 ROLE_TEMPLATES = (
@@ -73,8 +100,19 @@ def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
     from govoplan_notifications.backend.db.models import NotificationMessage
 
     return {
-        "notifications": session.query(NotificationMessage).filter(NotificationMessage.tenant_id == tenant_id, NotificationMessage.deleted_at.is_(None)).count(),
+        "notifications": session.query(NotificationMessage)
-        "pending_notifications": session.query(NotificationMessage).filter(NotificationMessage.tenant_id == tenant_id, NotificationMessage.status.in_(("pending", "queued", "failed")), NotificationMessage.deleted_at.is_(None)).count(),
+        .filter(
+            NotificationMessage.tenant_id == tenant_id,
+            NotificationMessage.deleted_at.is_(None),
+        )
+        .count(),
+        "pending_notifications": session.query(NotificationMessage)
+        .filter(
+            NotificationMessage.tenant_id == tenant_id,
+            NotificationMessage.status.in_(("pending", "queued", "failed")),
+            NotificationMessage.deleted_at.is_(None),
+        )
+        .count(),
     }
 
 
@@ -84,18 +122,89 @@ def _notifications_router(_context: ModuleContext):
     return router
 
 
+def _dsar_provider(_context: ModuleContext) -> NotificationsDsarProvider:
+    return NotificationsDsarProvider()
+
+
 manifest = ModuleManifest(
     id=MODULE_ID,
     name=MODULE_NAME,
     version=MODULE_VERSION,
-    required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
+    required_capabilities=(
-    optional_dependencies=("mail", "tasks", "portal", "workflow_engine", "calendar", "scheduling"),
+        CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
-    provides_interfaces=(ModuleInterfaceProvider(name="notifications.dispatch", version="0.1.8"),),
+        CAPABILITY_AUTH_PERMISSION_EVALUATOR,
+    ),
+    optional_dependencies=(
+        "mail",
+        "tasks",
+        "portal",
+        "workflow_engine",
+        "calendar",
+        "scheduling",
+    ),
+    provides_interfaces=(
+        ModuleInterfaceProvider(name="notifications.dispatch", version="0.1.8"),
+        ModuleInterfaceProvider(
+            name=NOTIFICATIONS_DSAR_CAPABILITY,
+            version="0.1.0",
+        ),
+    ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
     route_factory=_notifications_router,
     tenant_summary_providers=(_tenant_summary,),
     documentation=(
+        DocumentationTopic(
+            id="notifications.data-subject-requests",
+            title="Notification data-subject requests",
+            summary=(
+                "Export subject-addressed notifications and preferences, delete "
+                "personal inbox copies, and review external delivery evidence."
+            ),
+            body=(
+                "Notifications correlates verified account, membership, identity, "
+                "user, and email selectors only inside the active tenant. The access "
+                "package contains bounded message content, recipient fields, lifecycle "
+                "timestamps, source references, payload and metadata, and sanitized "
+                "delivery attempts. Personal badge, email, digest, and source-muting "
+                "preferences are also included. Explicit notification or preference "
+                "references can narrow a request and must still match the subject. "
+                "In-product inbox messages are derived copies, so erasure can delete "
+                "them after recipient and change checks without following the source "
+                "reference into its owning module. Personal preferences can also be "
+                "deleted and then return to defaults. Mail and other external-channel "
+                "records require manual retention review because provider acceptance "
+                "and delivery evidence may remain outside Notifications. Repeated "
+                "execution is unchanged; unrelated tenants and recipients are never "
+                "included."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "tenant_admin", "operator", "auditor"),
+            related_modules=("core", "mail", "tasks", "workflow_engine"),
+            metadata={
+                "help_contexts": [
+                    "notifications.page.inbox",
+                    "notifications.page.detail",
+                    "notifications.page.delivery",
+                    "notifications.settings.preferences",
+                    "privacy.data-subject-requests",
+                ],
+                "consequence_classes": {
+                    "delete_inbox_copy": (
+                        "Removes the derived in-product notification and attempts, "
+                        "not the source-module record."
+                    ),
+                    "delete_preferences": (
+                        "Removes personal overrides; notification defaults apply again."
+                    ),
+                    "review_external_delivery": (
+                        "Requires an authorized retention review before changing "
+                        "provider-delivery evidence."
+                    ),
+                },
+            },
+        ),
         DocumentationTopic(
             id="notifications.center-and-preferences",
             title="Use the notification center",
@@ -266,6 +375,17 @@ manifest = ModuleManifest(
             "govoplan_notifications.backend.capabilities",
             fromlist=["dispatch_capability"],
         ).dispatch_capability(context),
+        NOTIFICATIONS_DSAR_CAPABILITY: _dsar_provider,
+    },
+    capability_documentation={
+        NOTIFICATIONS_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Notifications data-subject request provider",
+            summary=(
+                "Exports subject-addressed notification data, deletes personal "
+                "inbox copies and preferences, and isolates external delivery evidence."
+            ),
+            contract_version="0.1.0",
+        ),
     },
     architecture=declared_module_architecture(
         layer="communication_participation",
@@ -273,8 +393,14 @@ manifest = ModuleManifest(
         maturity="vertical_slice",
         documentation_ref="docs/NOTIFICATION_INBOX_BOUNDARY.md",
         test_ref="tests/test_notifications.py",
-        known_limits=("Production email delivery depends on the optional Mail capability and does not provide an independent transport.",),
+        known_limits=(
-        owned_concepts=("notification", "notification preference", "notification delivery attempt"),
+            "Production email delivery depends on the optional Mail capability and does not provide an independent transport.",
+        ),
+        owned_concepts=(
+            "notification",
+            "notification preference",
+            "notification delivery attempt",
+        ),
         non_owned_concepts=("mail transport", "domain event", "portal message"),
         recovery_docs=("docs/EMAIL_DELIVERY.md",),
         operations_docs=("docs/EMAIL_DELIVERY.md",),

tests/test_dsar_provider.py

@@ -0,0 +1,476 @@
+from __future__ import annotations
+
+import json
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarProvider,
+    DsarRecordRef,
+    DsarSubjectRef,
+)
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    create_data_subject_request,
+    search_data_subject_request,
+)
+from govoplan_notifications.backend.db.models import (
+    NotificationDeliveryAttempt,
+    NotificationMessage,
+    NotificationPreference,
+)
+from govoplan_notifications.backend.dsar_provider import (
+    NOTIFICATIONS_DSAR_CAPABILITY,
+    NotificationsDsarProvider,
+)
+from govoplan_notifications.backend.manifest import manifest
+
+
+class _Registry:
+    def __init__(
+        self,
+        provider: NotificationsDsarProvider,
+        *,
+        active: bool = True,
+    ) -> None:
+        self.provider = provider
+        self.active = active
+
+    def capability_names(self):
+        return (NOTIFICATIONS_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        self._assert_capability(name)
+        return "notifications"
+
+    def tenant_entitlement_resolver(self):
+        active = self.active
+
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type(
+                    "State",
+                    (),
+                    {"effective_modules": ("notifications",) if active else ()},
+                )()
+
+        return _Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del session, kwargs
+        self._assert_capability(name)
+        return self.provider
+
+    def manifests(self):
+        return (type("Manifest", (), {"id": "notifications"})(),)
+
+    @staticmethod
+    def _assert_capability(name: str) -> None:
+        if name != NOTIFICATIONS_DSAR_CAPABILITY:
+            raise KeyError(name)
+
+
+class NotificationsDsarProviderTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        Base.metadata.create_all(self.engine)
+        self.session = Session(self.engine)
+        self.provider = NotificationsDsarProvider()
+        self.assertIsInstance(self.provider, DsarProvider)
+        self._seed()
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def _message(
+        self,
+        notification_id: str,
+        *,
+        tenant_id: str = "tenant-1",
+        channel: str = "inbox",
+        recipient: str | None = None,
+        recipient_id: str | None = "membership-1",
+        subject: str,
+    ) -> NotificationMessage:
+        return NotificationMessage(
+            id=notification_id,
+            tenant_id=tenant_id,
+            source_module="tasks",
+            source_resource_type="task",
+            source_resource_id=f"task-{notification_id}",
+            event_kind="task_due",
+            channel=channel,
+            recipient=recipient,
+            recipient_type="email" if recipient else "membership",
+            recipient_id=recipient_id,
+            recipient_label="Resident Example",
+            subject=subject,
+            body_text=f"Private body for {notification_id}",
+            body_html=f"<p>Private HTML for {notification_id}</p>",
+            action_url="/tasks",
+            priority=2,
+            status="sent",
+            attempt_count=0,
+            payload={
+                "case_reference": f"CASE-{notification_id}",
+                "access_token": "provider-secret-do-not-export",
+            },
+            metadata_={"classification": "personal"},
+        )
+
+    def _seed(self) -> None:
+        inbox = self._message(
+            "notification-inbox",
+            subject="Personal inbox notice",
+        )
+        mail = self._message(
+            "notification-mail",
+            channel="mail",
+            recipient="Resident@Example.test",
+            recipient_id=None,
+            subject="Personal mail notice",
+        )
+        mail.attempt_count = 1
+        mail.attempts.append(
+            NotificationDeliveryAttempt(
+                id="attempt-mail",
+                tenant_id="tenant-1",
+                attempt_no=1,
+                channel="mail",
+                provider="mail.delivery_outbox",
+                status="accepted",
+                external_message_id="mail-command-1",
+                details={
+                    "provider_state": "accepted",
+                    "authorization": "Bearer provider-secret-do-not-export",
+                },
+            )
+        )
+        self.session.add_all(
+            (
+                inbox,
+                mail,
+                self._message(
+                    "notification-other-recipient",
+                    recipient_id="membership-other",
+                    subject="Other recipient private notice",
+                ),
+                self._message(
+                    "notification-other-tenant",
+                    tenant_id="tenant-2",
+                    subject="Other tenant private notice",
+                ),
+                NotificationPreference(
+                    id="preference-personal",
+                    tenant_id="tenant-1",
+                    user_id="membership-1",
+                    show_unread_badge=False,
+                    email_enabled=True,
+                    email_digest_enabled=True,
+                    muted_source_modules=["campaign", "calendar"],
+                    metadata_={"digest_hour": 8},
+                ),
+                NotificationPreference(
+                    id="preference-other",
+                    tenant_id="tenant-1",
+                    user_id="membership-other",
+                    show_unread_badge=True,
+                    email_enabled=False,
+                    email_digest_enabled=False,
+                    muted_source_modules=[],
+                    metadata_={},
+                ),
+            )
+        )
+
+    @staticmethod
+    def _subject() -> DsarSubjectRef:
+        return DsarSubjectRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            identity_id="identity-1",
+            email="resident@example.test",
+        )
+
+    def test_search_is_tenant_recipient_scoped_and_exports_bounded_content(
+        self,
+    ) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self._subject(),
+        )
+
+        self.assertEqual(
+            [
+                "preference-personal",
+                "notification-inbox",
+                "notification-mail",
+            ],
+            [record.resource_id for record in records],
+        )
+        exported = json.dumps([record.to_dict() for record in records])
+        self.assertIn("Private body for notification-inbox", exported)
+        self.assertIn("CASE-notification-mail", exported)
+        self.assertIn("mail-command-1", exported)
+        self.assertIn("digest_hour", exported)
+        self.assertIn("[redacted]", exported)
+        self.assertNotIn("provider-secret-do-not-export", exported)
+        self.assertNotIn("notification-other-recipient", exported)
+        self.assertNotIn("notification-other-tenant", exported)
+
+    def test_resource_references_narrow_and_alias_conflicts_fail_closed(self) -> None:
+        notification = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                membership_id="membership-1",
+                external_references={
+                    "notifications.notification": "notification-inbox"
+                },
+            ),
+        )
+        preference = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                membership_id="membership-1",
+                external_references={"notifications.preference": "preference-personal"},
+            ),
+        )
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"notifications.account": "account-other"},
+            ),
+        )
+        reference_only = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                external_references={"notifications.notification": "notification-inbox"}
+            ),
+        )
+
+        self.assertEqual(
+            ["notification-inbox"], [item.resource_id for item in notification]
+        )
+        self.assertEqual(
+            ["preference-personal"], [item.resource_id for item in preference]
+        )
+        self.assertEqual((), conflict)
+        self.assertEqual((), reference_only)
+
+    def test_erasure_deletes_preferences_and_inbox_but_reviews_mail(self) -> None:
+        subject = self._subject()
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+        )
+        actions = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            records=records,
+        )
+        self.assertEqual(
+            ["delete", "delete", "manual_review"],
+            [action.kind for action in actions],
+        )
+
+        first = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=actions,
+            request_id="dsar-1",
+        )
+        second = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=actions,
+            request_id="dsar-1",
+        )
+
+        self.assertEqual(
+            ["executed", "executed", "blocked"],
+            [result.status for result in first],
+        )
+        self.assertEqual(
+            ["unchanged", "unchanged", "blocked"],
+            [result.status for result in second],
+        )
+        self.assertIsNone(
+            self.session.get(NotificationPreference, "preference-personal")
+        )
+        self.assertIsNone(self.session.get(NotificationMessage, "notification-inbox"))
+        self.assertIsNotNone(self.session.get(NotificationMessage, "notification-mail"))
+        self.assertIsNotNone(
+            self.session.get(NotificationMessage, "notification-other-recipient")
+        )
+
+    def test_changed_and_foreign_resources_are_blocked(self) -> None:
+        subject = self._subject()
+        record = next(
+            item
+            for item in self.provider.search_subject(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+            )
+            if item.resource_id == "notification-inbox"
+        )
+        action = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            records=(record,),
+        )[0]
+        notification = self.session.get(NotificationMessage, "notification-inbox")
+        notification.subject = "Changed after planning"
+        self.session.flush()
+        changed = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=(action,),
+            request_id="dsar-1",
+        )
+        self.assertEqual("blocked", changed[0].status)
+
+        with self.assertRaisesRegex(ValueError, "foreign provider record"):
+            self.provider.plan_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+                records=(
+                    DsarRecordRef(
+                        provider_id="mail",
+                        module_id="mail",
+                        resource_type="notification_message",
+                        resource_id="notification-inbox",
+                        category="message",
+                        title="Foreign message",
+                    ),
+                ),
+            )
+        with self.assertRaisesRegex(ValueError, "foreign provider action"):
+            self.provider.execute_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+                actions=(
+                    DsarErasureActionRef(
+                        action_id="mail:delete:notification:notification-inbox",
+                        provider_id="mail",
+                        module_id="mail",
+                        kind="delete",
+                        resource_type="notification_message",
+                        resource_id="notification-inbox",
+                        title="Delete notification",
+                        rationale="Foreign action",
+                        executable=True,
+                    ),
+                ),
+                request_id="dsar-1",
+            )
+
+    def test_oversized_content_fails_closed(self) -> None:
+        notification = self.session.get(
+            NotificationMessage,
+            "notification-inbox",
+        )
+        notification.body_text = "x" * 100_001
+        self.session.flush()
+
+        with self.assertRaisesRegex(ValueError, "body text exceeds"):
+            self.provider.search_subject(
+                self.session,
+                tenant_id="tenant-1",
+                subject=DsarSubjectRef(
+                    membership_id="membership-1",
+                    external_references={
+                        "notifications.notification": "notification-inbox"
+                    },
+                ),
+            )
+
+    def test_core_workflow_and_manifest_register_provider(self) -> None:
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-NOTIFICATIONS-1",
+            request_kind="access",
+            subject=self._subject(),
+            purpose="Respond to a verified request.",
+            legal_basis="Article 15 GDPR",
+            due_at=None,
+            requested_by_account_id="privacy-officer",
+        )
+        self.session.commit()
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider),
+            row=row,
+            expected_revision=1,
+        )
+        self.assertEqual(
+            [NOTIFICATIONS_DSAR_CAPABILITY],
+            row.coverage["provider_capabilities"],
+        )
+        self.assertEqual(3, row.search_result["record_count"])
+
+        inactive = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-NOTIFICATIONS-2",
+            request_kind="access",
+            subject=self._subject(),
+            purpose="Respond to a verified request.",
+            legal_basis="Article 15 GDPR",
+            due_at=None,
+            requested_by_account_id="privacy-officer",
+        )
+        self.session.commit()
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider, active=False),
+            row=inactive,
+            expected_revision=1,
+        )
+        self.assertEqual(
+            [NOTIFICATIONS_DSAR_CAPABILITY],
+            inactive.coverage["inactive_provider_capabilities"],
+        )
+
+        self.assertIn(NOTIFICATIONS_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(
+            NOTIFICATIONS_DSAR_CAPABILITY,
+            manifest.capability_documentation,
+        )
+        self.assertIn(
+            NOTIFICATIONS_DSAR_CAPABILITY,
+            {item.name for item in manifest.provides_interfaces},
+        )
+        self.assertTrue(
+            any(
+                topic.id == "notifications.data-subject-requests"
+                and {"admin", "user"}.issubset(topic.documentation_types)
+                for topic in manifest.documentation
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()