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

from __future__ import annotations

from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, Literal, cast

from sqlalchemy.orm import Session

from govoplan_core.admin.models import SystemSettings
from govoplan_core.admin.settings import SYSTEM_SETTINGS_ID
from govoplan_core.core.policy import (
    PolicySourceStep,
    SchedulingParticipantPrivacyDecision,
    SchedulingParticipantPrivacyRequest,
    SchedulingParticipantVisibility,
)
from govoplan_core.tenancy.scope import Tenant


SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY = "scheduling_participant_privacy_policy"
MAXIMUM_VISIBILITY_KEY = "maximum_visibility"

_AGGREGATES_ONLY: SchedulingParticipantVisibility = "aggregates_only"
_NAMES_AND_STATUSES: SchedulingParticipantVisibility = "names_and_statuses"


@dataclass(frozen=True, slots=True)
class _VisibilityCeiling:
    value: SchedulingParticipantVisibility
    configured: bool = False
    issue: str | None = None


def _visibility_ceiling(settings_payload: object) -> _VisibilityCeiling:
    """Read one ceiling, treating malformed explicit policy as restrictive."""

    if settings_payload is None:
        return _VisibilityCeiling(value=_NAMES_AND_STATUSES)
    if not isinstance(settings_payload, Mapping):
        return _VisibilityCeiling(
            value=_AGGREGATES_ONLY,
            configured=True,
            issue="invalid_settings_shape",
        )
    if SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY not in settings_payload:
        return _VisibilityCeiling(value=_NAMES_AND_STATUSES)

    raw_policy = settings_payload.get(SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY)
    if not isinstance(raw_policy, Mapping) or MAXIMUM_VISIBILITY_KEY not in raw_policy:
        return _VisibilityCeiling(
            value=_AGGREGATES_ONLY,
            configured=True,
            issue="invalid_policy_shape",
        )

    raw_visibility = raw_policy.get(MAXIMUM_VISIBILITY_KEY)
    if raw_visibility == _AGGREGATES_ONLY:
        return _VisibilityCeiling(value=_AGGREGATES_ONLY, configured=True)
    if raw_visibility == _NAMES_AND_STATUSES:
        return _VisibilityCeiling(value=_NAMES_AND_STATUSES, configured=True)
    return _VisibilityCeiling(
        value=_AGGREGATES_ONLY,
        configured=True,
        issue="invalid_maximum_visibility",
    )


def _restrict_visibility(
    current: SchedulingParticipantVisibility,
    ceiling: SchedulingParticipantVisibility,
) -> SchedulingParticipantVisibility:
    if current == _AGGREGATES_ONLY or ceiling == _AGGREGATES_ONLY:
        return _AGGREGATES_ONLY
    return _NAMES_AND_STATUSES


def _source_step(
    *,
    scope_type: Literal["system", "tenant"],
    scope_id: str | None,
    label: str,
    ceiling: _VisibilityCeiling,
) -> PolicySourceStep | None:
    if not ceiling.configured:
        return None
    policy: dict[str, Any] = {MAXIMUM_VISIBILITY_KEY: ceiling.value}
    if ceiling.issue is not None:
        policy["configuration_status"] = "invalid_fail_closed"
    return PolicySourceStep(
        scope_type=scope_type,
        scope_id=scope_id,
        label=label,
        applied_fields=(MAXIMUM_VISIBILITY_KEY,),
        policy=policy,
    )


def _invalid_requested_visibility(value: object) -> bool:
    return value not in {_AGGREGATES_ONLY, _NAMES_AND_STATUSES}


class SqlSchedulingParticipantPrivacyPolicy:
    """Resolve Scheduling roster disclosure against system and tenant ceilings.

    Scheduling owns the request-level visibility setting. This provider is an
    optional upper bound: absent policy preserves that setting, while an
    explicit or malformed policy may only reduce it.
    """

    def resolve_scheduling_participant_visibility(
        self,
        session: object,
        *,
        request: SchedulingParticipantPrivacyRequest,
    ) -> SchedulingParticipantPrivacyDecision:
        db_session = cast(Session, session)
        configuration_errors: list[dict[str, str]] = []

        requested_invalid = _invalid_requested_visibility(request.requested_visibility)
        requested_visibility: SchedulingParticipantVisibility = (
            _AGGREGATES_ONLY if requested_invalid else request.requested_visibility
        )
        if requested_invalid:
            configuration_errors.append({"scope": "request", "code": "invalid_requested_visibility"})

        system_settings = db_session.get(SystemSettings, SYSTEM_SETTINGS_ID)
        system_ceiling = _visibility_ceiling(system_settings.settings if system_settings is not None else None)
        if system_ceiling.issue is not None:
            configuration_errors.append({"scope": "system", "code": system_ceiling.issue})

        tenant = db_session.get(Tenant, request.tenant_id)
        if tenant is None:
            tenant_ceiling = _VisibilityCeiling(
                value=_AGGREGATES_ONLY,
                configured=True,
                issue="tenant_not_found",
            )
        else:
            tenant_ceiling = _visibility_ceiling(tenant.settings)
        if tenant_ceiling.issue is not None:
            configuration_errors.append({"scope": "tenant", "code": tenant_ceiling.issue})

        policy_ceiling = _restrict_visibility(system_ceiling.value, tenant_ceiling.value)
        effective_visibility = _restrict_visibility(requested_visibility, policy_ceiling)

        source_path = tuple(
            step
            for step in (
                _source_step(
                    scope_type="system",
                    scope_id=None,
                    label="System",
                    ceiling=system_ceiling,
                ),
                _source_step(
                    scope_type="tenant",
                    scope_id=request.tenant_id,
                    label="Tenant",
                    ceiling=tenant_ceiling,
                ),
            )
            if step is not None
        )

        reason: str | None = None
        if configuration_errors:
            reason = (
                "Participant roster visibility was restricted because policy configuration "
                "could not be validated."
            )
        elif effective_visibility != requested_visibility:
            reason = "Participant roster visibility is restricted by policy."

        return SchedulingParticipantPrivacyDecision(
            effective_visibility=effective_visibility,
            reason=reason,
            source_path=source_path,
            details={
                "requested_visibility": request.requested_visibility,
                "policy_ceiling": policy_ceiling,
                "configured_scopes": [
                    scope
                    for scope, ceiling in (("system", system_ceiling), ("tenant", tenant_ceiling))
                    if ceiling.configured
                ],
                "configuration_errors": configuration_errors,
            },
        )


__all__ = [
    "MAXIMUM_VISIBILITY_KEY",
    "SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY",
    "SqlSchedulingParticipantPrivacyPolicy",
]