govoplan-helpdesk/src/govoplan_helpdesk/backend/dsar_provider.py

from __future__ import annotations

from collections.abc import Sequence
from datetime import datetime, timezone

from sqlalchemy.orm import Session

from govoplan_core.core.dsar import (
    DsarErasureActionRef,
    DsarExecutionResultRef,
    DsarRecordRef,
    DsarSubjectRef,
    dsar_capability_name,
)
from govoplan_helpdesk.backend.db.models import HelpdeskProfileHistory, HelpdeskServiceProfile


HELPDESK_DSAR_CAPABILITY = dsar_capability_name("helpdesk")
_CONFLICT = object()


class HelpdeskDsarProvider:
    provider_id = "helpdesk"
    module_id = "helpdesk"

    def search_subject(self, session: object, *, tenant_id: str, subject: DsarSubjectRef) -> Sequence[DsarRecordRef]:
        db = _session(session)
        actor_ids = _actor_ids(subject)
        if actor_ids is None:
            return ()
        rows = (
            db.query(HelpdeskProfileHistory, HelpdeskServiceProfile)
            .join(HelpdeskServiceProfile, HelpdeskProfileHistory.profile_id == HelpdeskServiceProfile.id)
            .filter(
                HelpdeskProfileHistory.tenant_id == tenant_id,
                HelpdeskProfileHistory.actor_id.in_(actor_ids),
            )
            .order_by(HelpdeskProfileHistory.occurred_at.asc())
            .limit(5_001)
            .all()
        )
        if len(rows) > 5_000:
            raise ValueError("Helpdesk DSAR attribution limit exceeded; narrow the selectors.")
        return tuple(
            DsarRecordRef(
                provider_id="helpdesk",
                module_id="helpdesk",
                resource_type="service_profile_actor_attribution",
                resource_id=history.id,
                category="configuration_accountability_evidence",
                title=f"Helpdesk profile lifecycle attribution: {profile.profile_key}",
                data={
                    "activity": "configured_helpdesk_service_profile",
                    "profile_id": profile.id,
                    "profile_key": profile.profile_key,
                    "revision": history.revision,
                    "occurred_at": _iso(history.occurred_at),
                },
                observed_at=_aware(history.occurred_at),
                immutable_evidence=True,
                retention_reason="Helpdesk service-profile change attribution is immutable configuration accountability evidence.",
            )
            for history, profile in rows
        )

    def plan_erasure(self, session: object, *, tenant_id: str, subject: DsarSubjectRef, records: Sequence[DsarRecordRef]) -> Sequence[DsarErasureActionRef]:
        del tenant_id
        _session(session)
        if _actor_ids(subject) is None:
            raise ValueError("Helpdesk DSAR subject selectors conflict.")
        return tuple(
            DsarErasureActionRef(
                action_id=f"helpdesk:retain:{record.resource_id}",
                provider_id="helpdesk",
                module_id="helpdesk",
                kind="retain",
                resource_type=record.resource_type,
                resource_id=record.resource_id,
                title=f"Retain {record.title}",
                rationale=record.retention_reason or "Configuration attribution is immutable evidence.",
                executable=False,
            )
            for record in records
            if _valid_record(record)
        )

    def execute_erasure(self, session: object, *, tenant_id: str, subject: DsarSubjectRef, actions: Sequence[DsarErasureActionRef], request_id: str) -> Sequence[DsarExecutionResultRef]:
        del tenant_id
        _session(session)
        if _actor_ids(subject) is None:
            raise ValueError("Helpdesk DSAR subject selectors conflict.")
        results = []
        for action in actions:
            if action.provider_id != "helpdesk" or action.module_id != "helpdesk" or action.kind != "retain" or action.executable:
                raise ValueError("Helpdesk DSAR publishes retain-only actions.")
            results.append(
                DsarExecutionResultRef(
                    action_id=action.action_id,
                    status="blocked",
                    summary="Helpdesk configuration attribution remains immutable evidence.",
                    evidence={"request_id": request_id},
                )
            )
        return tuple(results)


def _actor_ids(subject: DsarSubjectRef) -> tuple[str, ...] | None:
    refs = subject.external_references
    values = (
        _coalesce(subject.account_id, refs.get("helpdesk.account"), refs.get("access.account")),
        _coalesce(subject.identity_id, refs.get("helpdesk.identity"), refs.get("identity.id")),
        _coalesce(subject.membership_id, refs.get("helpdesk.membership"), refs.get("tenancy.membership")),
    )
    if any(value is _CONFLICT for value in values):
        return None
    result = tuple(dict.fromkeys(value for value in values if isinstance(value, str) and value))
    return result or None


def _coalesce(*values: str | None) -> str | None | object:
    normalized = {str(value).strip() for value in values if str(value or "").strip()}
    return _CONFLICT if len(normalized) > 1 else next(iter(normalized), None)


def _valid_record(record: DsarRecordRef) -> bool:
    if record.provider_id != "helpdesk" or record.module_id != "helpdesk" or record.resource_type != "service_profile_actor_attribution":
        raise ValueError("Helpdesk DSAR record identity is invalid.")
    return True


def _aware(value: datetime | None) -> datetime | None:
    if value is not None and value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value


def _iso(value: datetime | None) -> str | None:
    result = _aware(value)
    return result.isoformat() if result else None


def _session(value: object) -> Session:
    if not isinstance(value, Session):
        raise TypeError("Helpdesk DSAR requires a SQLAlchemy Session.")
    return value


__all__ = ["HELPDESK_DSAR_CAPABILITY", "HelpdeskDsarProvider"]