govoplan-views/src/govoplan_views/backend/impact_subjects.py

from __future__ import annotations

from collections.abc import Mapping
from typing import Any, cast

from sqlalchemy import or_
from sqlalchemy.orm import Session

from govoplan_core.core.policy import (
    POLICY_SCOPE_TYPES,
    PolicyImpactPopulationRequest,
    PolicyImpactSubject,
    PolicyImpactSubjectBatch,
    PolicyScopeType,
)
from govoplan_views.backend.db.models import ViewDefinition


VIEW_IMPACT_ACTIONS = (
    "view",
    "select",
    "assign",
    "edit",
    "derive",
    "workflow_activate",
)


class ViewsPolicyImpactSubjectProvider:
    provider_id = "views"
    supported_policy_families = ("view",)

    def __init__(self, registry: object) -> None:
        self._registry = registry

    def collect_policy_impact_subjects(
        self,
        session: object | None = None,
        *,
        request: PolicyImpactPopulationRequest,
    ) -> PolicyImpactSubjectBatch:
        if request.policy_family != "view":
            return PolicyImpactSubjectBatch(
                provider_id=self.provider_id,
                state="unavailable",
                explanation="Views only contributes subjects for the View policy family.",
            )
        if not isinstance(session, Session):
            return PolicyImpactSubjectBatch(
                provider_id=self.provider_id,
                state="unavailable",
                explanation="The Views impact population requires a database session.",
            )

        selector = request.selector
        _validate_selector_keys(selector)
        actions = _actions(selector)
        include_views = _selector_flag(selector, "include_views", default=True)
        include_surfaces = _selector_flag(
            selector,
            "include_surfaces",
            default=True,
        )
        explicit_view_ids = _selector_ids(selector, "view_ids")
        explicit_surface_ids = _selector_ids(selector, "surface_ids")

        subjects: list[PolicyImpactSubject] = []
        total_available = 0
        if include_views:
            query = session.query(ViewDefinition).filter(
                ViewDefinition.deleted_at.is_(None),
                or_(
                    ViewDefinition.tenant_id == request.tenant_id,
                    ViewDefinition.tenant_id.is_(None),
                ),
            )
            if explicit_view_ids is not None:
                query = query.filter(ViewDefinition.id.in_(explicit_view_ids))
            definition_count = query.count()
            definition_limit = max(
                1,
                (request.limit + len(actions) - 1) // len(actions),
            )
            definitions = query.order_by(
                ViewDefinition.scope_type.asc(),
                ViewDefinition.name.asc(),
                ViewDefinition.id.asc(),
            ).limit(definition_limit).all()
            total_available += definition_count * len(actions)
            for definition in definitions:
                scope_type = (
                    cast(PolicyScopeType, definition.scope_type)
                    if definition.scope_type in POLICY_SCOPE_TYPES
                    else None
                )
                for action in actions:
                    subjects.append(
                        PolicyImpactSubject(
                            module_id="views",
                            resource_type="view",
                            resource_id=definition.id,
                            action=action,
                            label=(
                                definition.name
                                if request.allow_sensitive_details
                                else None
                            ),
                            scope_type=scope_type,
                            scope_id=definition.scope_id,
                            attributes={"definition_scope": definition.scope_type},
                        )
                    )

        if include_surfaces:
            surfaces = _surfaces(self._registry)
            if explicit_surface_ids is not None:
                allowed_surface_ids = set(explicit_surface_ids)
                surfaces = tuple(
                    surface
                    for surface in surfaces
                    if str(getattr(surface, "id", "")) in allowed_surface_ids
                )
            total_available += len(surfaces)
            for surface in surfaces:
                subjects.append(
                    PolicyImpactSubject(
                        module_id=str(getattr(surface, "module_id", "core")),
                        resource_type="surface",
                        resource_id=str(getattr(surface, "id", "")),
                        action="view",
                        label=(
                            str(getattr(surface, "label", "")) or None
                            if request.allow_sensitive_details
                            else None
                        ),
                        attributes={"surface_kind": str(getattr(surface, "kind", ""))},
                    )
                )

        subjects = subjects[: request.limit]
        truncated = total_available > len(subjects)
        return PolicyImpactSubjectBatch(
            provider_id=self.provider_id,
            subjects=tuple(subjects),
            state="truncated" if truncated else "complete",
            total_available=total_available,
            explanation=(
                f"The bounded preview returned {len(subjects)} of "
                f"{total_available} matching View subjects."
                if truncated
                else "The explicit Views catalogue population was evaluated completely."
            ),
        )


def _actions(selector: Mapping[str, Any]) -> tuple[str, ...]:
    value = selector.get("actions")
    if value is None:
        return VIEW_IMPACT_ACTIONS
    if not isinstance(value, (list, tuple)):
        raise ValueError("View impact actions must be a list")
    actions = tuple(dict.fromkeys(str(item).strip() for item in value))
    if not actions or any(action not in VIEW_IMPACT_ACTIONS for action in actions):
        raise ValueError("View impact actions contain an unsupported action")
    return actions


def _validate_selector_keys(selector: Mapping[str, Any]) -> None:
    supported = {
        "actions",
        "include_views",
        "include_surfaces",
        "view_ids",
        "surface_ids",
    }
    unknown = sorted(str(key) for key in selector if str(key) not in supported)
    if unknown:
        raise ValueError(
            "View impact selector contains unsupported fields: " + ", ".join(unknown)
        )


def _selector_flag(
    selector: Mapping[str, Any],
    key: str,
    *,
    default: bool,
) -> bool:
    value = selector.get(key, default)
    if not isinstance(value, bool):
        raise ValueError(f"View impact selector {key} must be boolean")
    return value


def _selector_ids(
    selector: Mapping[str, Any],
    key: str,
) -> tuple[str, ...] | None:
    if key not in selector:
        return None
    value = selector.get(key)
    if not isinstance(value, (list, tuple)) or len(value) > 500:
        raise ValueError(f"View impact selector {key} must contain at most 500 IDs")
    result = tuple(dict.fromkeys(str(item).strip() for item in value))
    if any(not item or len(item) > 240 for item in result):
        raise ValueError(f"View impact selector {key} contains an invalid ID")
    return result


def _surfaces(registry: object) -> tuple[object, ...]:
    if not hasattr(registry, "view_surfaces"):
        return ()
    value = registry.view_surfaces()
    return tuple(value) if isinstance(value, (list, tuple)) else tuple(value or ())


__all__ = ["VIEW_IMPACT_ACTIONS", "ViewsPolicyImpactSubjectProvider"]