govoplan-quick-access/src/govoplan_quick_access/backend/dsar_provider.py

from __future__ import annotations

from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone

from sqlalchemy import or_
from sqlalchemy.orm import Session

from govoplan_core.core.dsar import (
    DsarErasureActionRef,
    DsarExecutionResultRef,
    DsarRecordRef,
    DsarSubjectRef,
    dsar_capability_name,
)
from govoplan_quick_access.backend.db.models import QuickAccessProfile


QUICK_ACCESS_DSAR_CAPABILITY = dsar_capability_name("quick_access")
_MAX_PREFERENCES = 1_000
_CONFLICT = object()


@dataclass(frozen=True, slots=True)
class _SubjectSelectors:
    account_id: str
    profile_id: str | None


class QuickAccessDsarProvider:
    provider_id = "quick_access"
    module_id = "quick_access"

    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 ()

        personal_query = db.query(QuickAccessProfile).filter(
            QuickAccessProfile.scope_type == "user",
            QuickAccessProfile.tenant_id == tenant_id,
            QuickAccessProfile.scope_id == selectors.account_id,
        )
        attribution_query = db.query(QuickAccessProfile).filter(
            QuickAccessProfile.scope_type == "tenant",
            QuickAccessProfile.tenant_id == tenant_id,
            or_(
                QuickAccessProfile.created_by == selectors.account_id,
                QuickAccessProfile.updated_by == selectors.account_id,
            ),
        )
        if selectors.profile_id:
            personal_query = personal_query.filter(
                QuickAccessProfile.id == selectors.profile_id
            )
            attribution_query = attribution_query.filter(
                QuickAccessProfile.id == selectors.profile_id
            )

        records = [_personal_record(row) for row in personal_query.all()]
        records.extend(
            _attribution_record(row, selectors.account_id)
            for row in attribution_query.all()
        )
        return tuple(
            sorted(records, key=lambda item: (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("Quick Access DSAR requires one corroborated account.")

        actions: list[DsarErasureActionRef] = []
        for record in records:
            _validate_record(record)
            if record.resource_type == "quick_access_tenant_attribution":
                actions.append(
                    DsarErasureActionRef(
                        action_id=(
                            "quick_access:retain:quick_access_tenant_attribution:"
                            f"{record.resource_id}"
                        ),
                        provider_id=self.provider_id,
                        module_id=self.module_id,
                        kind="retain",
                        resource_type=record.resource_type,
                        resource_id=record.resource_id,
                        title=f"Retain {record.title}",
                        rationale=record.retention_reason
                        or "Tenant Quick Access policy attribution is accountability evidence.",
                        executable=False,
                    )
                )
                continue

            profile = (
                db.query(QuickAccessProfile)
                .filter(
                    QuickAccessProfile.id == record.resource_id,
                    QuickAccessProfile.scope_type == "user",
                    QuickAccessProfile.tenant_id == tenant_id,
                )
                .one_or_none()
            )
            if (
                profile is None
                or profile.scope_id != selectors.account_id
                or (selectors.profile_id and profile.id != selectors.profile_id)
            ):
                actions.append(_manual_action(record))
                continue
            actions.append(
                DsarErasureActionRef(
                    action_id=(
                        "quick_access:delete:quick_access_personal_profile:"
                        f"{profile.id}:r{profile.revision}"
                    ),
                    provider_id=self.provider_id,
                    module_id=self.module_id,
                    kind="delete",
                    resource_type="quick_access_personal_profile",
                    resource_id=profile.id,
                    title="Delete personal Quick Access profile",
                    rationale=(
                        "The profile contains only account-owned presentation "
                        "preferences and can be recreated from governed defaults."
                    ),
                    executable=True,
                    irreversible=True,
                    metadata={
                        "account_id": profile.scope_id,
                        "revision": profile.revision,
                    },
                )
            )
        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("Quick Access DSAR requires one corroborated account.")

        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=(
                            "Tenant Quick Access policy attribution remains "
                            "institutional accountability evidence."
                        ),
                    )
                )
                continue
            if action.resource_type != "quick_access_personal_profile":
                raise ValueError("Unsupported executable Quick Access DSAR action.")

            profile = (
                db.query(QuickAccessProfile)
                .filter(
                    QuickAccessProfile.id == action.resource_id,
                    QuickAccessProfile.scope_type == "user",
                    QuickAccessProfile.tenant_id == tenant_id,
                )
                .one_or_none()
            )
            if profile is None:
                results.append(
                    DsarExecutionResultRef(
                        action_id=action.action_id,
                        status="unchanged",
                        summary="The personal Quick Access profile was already absent.",
                        evidence={"request_id": request_id},
                    )
                )
                continue

            expected_account = str(action.metadata.get("account_id") or "")
            expected_revision = action.metadata.get("revision")
            if (
                profile.scope_id != selectors.account_id
                or expected_account != selectors.account_id
                or (selectors.profile_id and profile.id != selectors.profile_id)
            ):
                results.append(
                    DsarExecutionResultRef(
                        action_id=action.action_id,
                        status="blocked",
                        summary=(
                            "The profile is not owned by the corroborated subject "
                            "account and selector context."
                        ),
                    )
                )
                continue
            if expected_revision != profile.revision:
                results.append(
                    DsarExecutionResultRef(
                        action_id=action.action_id,
                        status="blocked",
                        summary=(
                            "The profile changed after the erasure plan; create a "
                            "new plan before deleting it."
                        ),
                    )
                )
                continue

            profile_id = profile.id
            db.delete(profile)
            db.flush()
            results.append(
                DsarExecutionResultRef(
                    action_id=action.action_id,
                    status="executed",
                    summary=(
                        "Deleted the personal Quick Access profile without "
                        "changing tenant policy, tools, or domain data."
                    ),
                    evidence={"request_id": request_id, "profile_id": profile_id},
                )
            )
        return tuple(results)


def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
    references = subject.external_references
    account_id = _coalesce(
        subject.account_id,
        references.get("quick_access.account"),
        references.get("access.account"),
    )
    profile_id = _coalesce(
        references.get("quick_access.profile"),
        references.get("quick_access.profile_id"),
    )
    if _CONFLICT in (account_id, profile_id):
        return None
    normalized_account = _optional_string(account_id)
    if normalized_account is None:
        return None
    return _SubjectSelectors(
        account_id=normalized_account,
        profile_id=_optional_string(profile_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 _personal_record(profile: QuickAccessProfile) -> DsarRecordRef:
    return DsarRecordRef(
        provider_id="quick_access",
        module_id="quick_access",
        resource_type="quick_access_personal_profile",
        resource_id=profile.id,
        category="personal_interface_preference",
        title="Personal Quick Access profile",
        data={
            "revision": profile.revision,
            "category_preferences": _preference_projection(
                profile.category_preferences
            ),
            "tool_preferences": _preference_projection(profile.tool_preferences),
        },
        observed_at=_aware(profile.updated_at),
    )


def _attribution_record(
    profile: QuickAccessProfile,
    account_id: str,
) -> DsarRecordRef:
    activities = []
    if profile.created_by == account_id:
        activities.append("created_tenant_quick_access_policy")
    if profile.updated_by == account_id:
        activities.append("updated_tenant_quick_access_policy")
    return DsarRecordRef(
        provider_id="quick_access",
        module_id="quick_access",
        resource_type="quick_access_tenant_attribution",
        resource_id=profile.id,
        category="operator_accountability_evidence",
        title="Tenant Quick Access policy attribution",
        data={
            "activities": activities,
            "revision": profile.revision,
            "created_at": _iso(profile.created_at),
            "updated_at": _iso(profile.updated_at),
        },
        observed_at=_aware(profile.updated_at),
        immutable_evidence=True,
        retention_reason=(
            "Tenant policy authorship is retained as institutional "
            "accountability evidence; policy preferences are excluded."
        ),
    )


def _preference_projection(value: object) -> dict[str, dict[str, object]]:
    if not isinstance(value, Mapping):
        raise ValueError("Quick Access preference payload must be an object.")
    if len(value) > _MAX_PREFERENCES:
        raise ValueError(
            "Quick Access preference limit exceeded; review the stored profile."
        )
    projected: dict[str, dict[str, object]] = {}
    for raw_key, raw_entry in value.items():
        key = str(raw_key)
        if not key or len(key) > 255 or not isinstance(raw_entry, Mapping):
            raise ValueError("Quick Access preference entry is invalid.")
        entry: dict[str, object] = {}
        if isinstance(raw_entry.get("enabled"), bool):
            entry["enabled"] = raw_entry["enabled"]
        if isinstance(raw_entry.get("forced"), bool):
            entry["forced"] = raw_entry["forced"]
        order = raw_entry.get("order")
        if isinstance(order, int) and not isinstance(order, bool):
            entry["order"] = order
        projected[key] = entry
    return projected


def _manual_action(record: DsarRecordRef) -> DsarErasureActionRef:
    return DsarErasureActionRef(
        action_id=f"quick_access:manual_review:{record.resource_type}:{record.resource_id}",
        provider_id="quick_access",
        module_id="quick_access",
        kind="manual_review",
        resource_type=record.resource_type,
        resource_id=record.resource_id,
        title=f"Review {record.title}",
        rationale=(
            "The selected profile is absent or no longer belongs to the exact "
            "tenant, account, and narrowing selectors."
        ),
        executable=False,
    )


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("Quick Access DSAR requires a SQLAlchemy Session.")
    return value


def _validate_record(record: DsarRecordRef) -> None:
    if record.provider_id != "quick_access" or record.module_id != "quick_access":
        raise ValueError("Quick Access DSAR cannot plan a foreign provider record.")
    if (
        record.resource_type
        not in {
            "quick_access_personal_profile",
            "quick_access_tenant_attribution",
        }
        or not record.resource_id
    ):
        raise ValueError("Quick Access DSAR record identity is invalid.")


def _validate_action(action: DsarErasureActionRef) -> None:
    if action.provider_id != "quick_access" or action.module_id != "quick_access":
        raise ValueError("Quick Access DSAR cannot execute a foreign provider action.")
    if not action.action_id.startswith("quick_access:"):
        raise ValueError("Quick Access DSAR action identity is invalid.")


__all__ = ["QUICK_ACCESS_DSAR_CAPABILITY", "QuickAccessDsarProvider"]