govoplan-policy/src/govoplan_policy/backend/definition_policy_service.py

from __future__ import annotations

from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any

from sqlalchemy.orm import Session

from govoplan_policy.backend.db.models import PolicyOverride
from govoplan_policy.backend.definition_governance import (
    DEFINITION_POLICY_FIELDS,
    DefinitionPolicyResolution,
    resolve_definition_policy_rows,
    validate_definition_policy,
)
from govoplan_policy.backend.policy_overrides import (
    delete_policy_override,
    get_policy_override,
    normalize_policy_scope,
    resolution_policy_overrides,
    set_policy_override,
)


class DefinitionPolicyError(ValueError):
    pass


@dataclass(frozen=True, slots=True)
class DefinitionPolicyState:
    row: PolicyOverride | None
    local_policy: Mapping[str, bool]
    effective: DefinitionPolicyResolution
    parent: DefinitionPolicyResolution


def definition_policy_state(
    session: Session,
    *,
    module_id: str,
    tenant_id: str,
    scope_type: str,
    scope_id: str | None = None,
) -> DefinitionPolicyState:
    _, clean_scope_id, _ = normalize_policy_scope(
        tenant_id=tenant_id,
        scope_type=scope_type,
        scope_id=scope_id,
    )
    row = get_policy_override(
        session,
        policy_family="definition",
        target_key=module_id,
        tenant_id=tenant_id,
        scope_type=scope_type,
        scope_id=clean_scope_id,
    )
    local_policy, malformed = validate_definition_policy(
        row.policy if row is not None else {}
    )
    if malformed:
        local_policy = {}
    rows = _rows_for_scope(
        session,
        module_id=module_id,
        tenant_id=tenant_id,
        scope_type=scope_type,
        scope_id=clean_scope_id,
    )
    rank = _scope_rank(scope_type)
    return DefinitionPolicyState(
        row=row,
        local_policy=local_policy,
        effective=resolve_definition_policy_rows(rows),
        parent=resolve_definition_policy_rows(
            tuple(item for item in rows if _scope_rank(item.scope_type) < rank)
        ),
    )


def save_definition_policy(
    session: Session,
    *,
    module_id: str,
    tenant_id: str,
    scope_type: str,
    scope_id: str | None,
    policy: object,
    actor_id: str | None,
) -> DefinitionPolicyState:
    clean_policy, malformed = validate_definition_policy(policy)
    if malformed:
        raise DefinitionPolicyError(
            "Definition policy fields must be known boolean values"
        )
    before = definition_policy_state(
        session,
        module_id=module_id,
        tenant_id=tenant_id,
        scope_type=scope_type,
        scope_id=scope_id,
    )
    broadened = sorted(
        field
        for field, value in clean_policy.items()
        if value and before.parent.limits.get(field) is False
    )
    if broadened:
        raise DefinitionPolicyError(
            "Lower-scope policy cannot broaden parent restrictions: "
            + ", ".join(broadened)
        )
    set_policy_override(
        session,
        policy_family="definition",
        target_key=module_id,
        tenant_id=tenant_id,
        scope_type=scope_type,
        scope_id=scope_id,
        policy=clean_policy,
        actor_id=actor_id,
    )
    _clear_resolution_cache(session)
    return definition_policy_state(
        session,
        module_id=module_id,
        tenant_id=tenant_id,
        scope_type=scope_type,
        scope_id=scope_id,
    )


def remove_definition_policy(
    session: Session,
    *,
    module_id: str,
    tenant_id: str,
    scope_type: str,
    scope_id: str | None,
) -> bool:
    row = get_policy_override(
        session,
        policy_family="definition",
        target_key=module_id,
        tenant_id=tenant_id,
        scope_type=scope_type,
        scope_id=scope_id,
    )
    if row is None:
        return False
    delete_policy_override(session, row)
    _clear_resolution_cache(session)
    return True


def _rows_for_scope(
    session: Session,
    *,
    module_id: str,
    tenant_id: str,
    scope_type: str,
    scope_id: str | None,
) -> tuple[PolicyOverride, ...]:
    clean_scope = scope_type.strip().casefold()
    rows = resolution_policy_overrides(
        session,
        policy_family="definition",
        target_keys=("*", module_id),
        tenant_id=tenant_id,
        group_ids=(scope_id,) if clean_scope == "group" and scope_id else (),
        user_ids=(scope_id,) if clean_scope == "user" and scope_id else (),
    )
    maximum_rank = _scope_rank(clean_scope)
    return tuple(row for row in rows if _scope_rank(row.scope_type) <= maximum_rank)


def _scope_rank(scope_type: str) -> int:
    try:
        return ("system", "tenant", "group", "user").index(
            scope_type.strip().casefold()
        )
    except ValueError as exc:
        raise DefinitionPolicyError(
            "Definition policy scope must be system, tenant, group, or user"
        ) from exc


def _clear_resolution_cache(session: Session) -> None:
    session.info.pop("govoplan_policy_override_resolution", None)


def definition_policy_response_payload(
    state: DefinitionPolicyState,
) -> dict[str, Any]:
    local_policy: Mapping[str, Any] = state.local_policy
    if state.row is not None:
        _, malformed = validate_definition_policy(state.row.policy)
        if malformed:
            local_policy = {"configuration_status": "invalid_fail_closed"}
    return {
        "id": state.row.id if state.row is not None else None,
        "revision": state.row.revision if state.row is not None else None,
        "policy": dict(local_policy),
        "effective_policy": dict(state.effective.limits),
        "parent_policy": dict(state.parent.limits),
        "source_path": [step.to_dict() for step in state.effective.source_path],
        "diagnostics": [dict(item) for item in state.effective.diagnostics],
    }


__all__ = [
    "DEFINITION_POLICY_FIELDS",
    "DefinitionPolicyError",
    "DefinitionPolicyState",
    "definition_policy_response_payload",
    "definition_policy_state",
    "remove_definition_policy",
    "save_definition_policy",
]