DataOps-platform/deployment/app/core/governance/responsibilities.py

"""Versioned RACI bindings for governed resources and device accountability."""

from __future__ import annotations

import json
import uuid
from dataclasses import asdict, dataclass
from typing import Any

from sqlalchemy import text

from app.core.common.identifiers import new_governance_uid

RESOURCE_TYPES = frozenset(
    {
        "business_domain",
        "device_asset",
        "device_ontology",
        "device_mapping",
        "device_quality",
        "fault_classification",
        "quality_issue",
        "organization",
        "data_asset",
        "semantic_term",
        "data_standard",
        "quality_policy",
        "data_product",
        "agent",
    }
)
DEVICE_RESOURCE_TYPES = frozenset(
    {
        "device_asset",
        "device_ontology",
        "device_mapping",
        "device_quality",
        "fault_classification",
    }
)
RESPONSIBILITY_ROLES = frozenset(
    {
        "organization_owner",
        "domain_owner",
        "data_steward",
        "data_architect",
        "asset_manager",
        "term_steward",
        "standard_owner",
        "quality_owner",
        "product_owner",
        "agent_owner",
    }
)
RACI_ROLES = frozenset({"responsible", "accountable", "consulted", "informed"})


class ResponsibilityError(RuntimeError):
    """Base error for responsibility-matrix operations."""


class ResponsibilityValidationError(ResponsibilityError):
    """The requested matrix violates its governed contract."""


class ResponsibilityConflict(ResponsibilityError):
    """The matrix changed after the caller loaded it."""


class ResponsibilityUserUnavailable(ResponsibilityError):
    """An assignment targets an unknown or disabled user."""


@dataclass(frozen=True)
class ResponsibilityAssignment:
    user_id: str
    responsibility_role: str
    raci_role: str


def _valid_uuid(value: Any, field: str) -> str:
    try:
        return str(uuid.UUID(str(value)))
    except (ValueError, TypeError, AttributeError) as exc:
        raise ResponsibilityValidationError(f"{field} must be a UUID") from exc


def validate_resource(resource_type: str, resource_uid: str | None = None) -> None:
    if resource_type not in RESOURCE_TYPES:
        raise ResponsibilityValidationError("unsupported resource type")
    if resource_uid is None:
        return
    value = str(resource_uid).strip()
    if not value or len(value) > 120:
        raise ResponsibilityValidationError("resource uid is invalid")


def validate_matrix(
    resource_type: str,
    assignments: list[dict[str, Any]],
) -> tuple[ResponsibilityAssignment, ...]:
    validate_resource(resource_type)
    if not isinstance(assignments, list):
        raise ResponsibilityValidationError("responsibility matrix must be an array")

    validated: list[ResponsibilityAssignment] = []
    identities: set[tuple[str, str, str]] = set()
    for raw in assignments:
        if not isinstance(raw, dict):
            raise ResponsibilityValidationError("assignment must be an object")
        user_id = _valid_uuid(raw.get("user_id"), "user id")
        responsibility_role = str(raw.get("responsibility_role") or "").strip()
        raci_role = str(raw.get("raci_role") or "").strip()
        if responsibility_role not in RESPONSIBILITY_ROLES:
            raise ResponsibilityValidationError("unsupported responsibility role")
        if raci_role not in RACI_ROLES:
            raise ResponsibilityValidationError("unsupported RACI role")
        identity = (user_id, responsibility_role, raci_role)
        if identity in identities:
            raise ResponsibilityValidationError("duplicate responsibility assignment")
        identities.add(identity)
        validated.append(
            ResponsibilityAssignment(
                user_id=user_id,
                responsibility_role=responsibility_role,
                raci_role=raci_role,
            )
        )

    if resource_type in DEVICE_RESOURCE_TYPES:
        accountable_asset_managers = [
            item
            for item in validated
            if item.responsibility_role == "asset_manager"
            and item.raci_role == "accountable"
        ]
        if len(accountable_asset_managers) != 1:
            raise ResponsibilityValidationError(
                "device scope requires exactly one accountable asset manager"
            )
    return tuple(validated)


class ResponsibilityService:
    def __init__(self, repository):
        self.repository = repository

    def get(self, resource_type: str, resource_uid: str) -> dict[str, Any]:
        validate_resource(resource_type, resource_uid)
        return self.repository.get(resource_type, resource_uid)

    def replace(
        self,
        *,
        resource_type: str,
        resource_uid: str,
        assignments: list[dict[str, Any]],
        expected_revision: int,
        actor_uid: str,
    ) -> dict[str, Any]:
        validate_resource(resource_type, resource_uid)
        actor_uid = _valid_uuid(actor_uid, "actor uid")
        try:
            revision = int(expected_revision)
        except (TypeError, ValueError) as exc:
            raise ResponsibilityValidationError("revision is invalid") from exc
        if revision < 0:
            raise ResponsibilityValidationError("revision is invalid")
        return self.repository.replace(
            resource_type=resource_type,
            resource_uid=str(resource_uid).strip(),
            assignments=validate_matrix(resource_type, assignments),
            expected_revision=revision,
            actor_uid=actor_uid,
        )


class SqlAlchemyResponsibilityRepository:
    def __init__(self, session):
        self.session = session

    @staticmethod
    def _assignment_dict(row) -> dict[str, Any]:
        return {
            "user_id": str(row["user_id"]),
            "username": row["username"],
            "display_name": row["display_name"],
            "responsibility_role": row["responsibility_role"],
            "raci_role": row["raci_role"],
        }

    def _assignments(self, scope_id: str) -> list[dict[str, Any]]:
        rows = (
            self.session.execute(
                text(
                    """
                    SELECT a.user_id::text AS user_id, u.username,
                           u.display_name, a.responsibility_role, a.raci_role
                    FROM public.governance_responsibility_assignments a
                    JOIN public.users u ON u.id = a.user_id
                    WHERE a.scope_id = CAST(:scope_id AS uuid)
                    ORDER BY a.raci_role, a.responsibility_role, u.username
                    """
                ),
                {"scope_id": scope_id},
            )
            .mappings()
            .all()
        )
        return [self._assignment_dict(row) for row in rows]

    def get(self, resource_type: str, resource_uid: str) -> dict[str, Any]:
        scope = (
            self.session.execute(
                text(
                    """
                    SELECT id::text AS id, revision
                    FROM public.governance_responsibility_scopes
                    WHERE resource_type = :resource_type
                      AND resource_uid = :resource_uid
                    """
                ),
                {
                    "resource_type": resource_type,
                    "resource_uid": resource_uid,
                },
            )
            .mappings()
            .one_or_none()
        )
        if scope is None:
            return {
                "resource_type": resource_type,
                "resource_uid": resource_uid,
                "revision": 0,
                "assignments": [],
            }
        return {
            "resource_type": resource_type,
            "resource_uid": resource_uid,
            "revision": int(scope["revision"]),
            "assignments": self._assignments(scope["id"]),
        }

    def replace(
        self,
        *,
        resource_type: str,
        resource_uid: str,
        assignments: tuple[ResponsibilityAssignment, ...],
        expected_revision: int,
        actor_uid: str,
    ) -> dict[str, Any]:
        lock_key = f"responsibility:{resource_type}:{resource_uid}"
        self.session.execute(
            text("SELECT pg_advisory_xact_lock(hashtext(:key))"),
            {"key": lock_key},
        )
        scope = (
            self.session.execute(
                text(
                    """
                    SELECT id::text AS id, revision
                    FROM public.governance_responsibility_scopes
                    WHERE resource_type = :resource_type
                      AND resource_uid = :resource_uid
                    FOR UPDATE
                    """
                ),
                {
                    "resource_type": resource_type,
                    "resource_uid": resource_uid,
                },
            )
            .mappings()
            .one_or_none()
        )
        if scope is None:
            if expected_revision != 0:
                raise ResponsibilityConflict("responsibility revision conflict")
            scope_id = new_governance_uid()
            self.session.execute(
                text(
                    """
                    INSERT INTO public.governance_responsibility_scopes (
                        id, resource_type, resource_uid, revision,
                        updated_by
                    ) VALUES (
                        CAST(:id AS uuid), :resource_type, :resource_uid, 0,
                        CAST(:actor_uid AS uuid)
                    )
                    """
                ),
                {
                    "id": scope_id,
                    "resource_type": resource_type,
                    "resource_uid": resource_uid,
                    "actor_uid": actor_uid,
                },
            )
            current_revision = 0
            before: list[dict[str, Any]] = []
        else:
            scope_id = str(scope["id"])
            current_revision = int(scope["revision"])
            if current_revision != expected_revision:
                raise ResponsibilityConflict("responsibility revision conflict")
            before = self._assignments(scope_id)

        user_ids = sorted({item.user_id for item in assignments})
        active_user_ids = {
            str(row[0])
            for row in self.session.execute(
                text(
                    """
                    SELECT id::text FROM public.users
                    WHERE status = 'active' AND id::text = ANY(:user_ids)
                    """
                ),
                {"user_ids": user_ids},
            )
        }
        if active_user_ids != set(user_ids):
            raise ResponsibilityUserUnavailable(
                "responsibility user is unknown or disabled"
            )

        self.session.execute(
            text(
                """
                DELETE FROM public.governance_responsibility_assignments
                WHERE scope_id = CAST(:scope_id AS uuid)
                """
            ),
            {"scope_id": scope_id},
        )
        for assignment in assignments:
            self.session.execute(
                text(
                    """
                    INSERT INTO public.governance_responsibility_assignments (
                        id, scope_id, user_id, responsibility_role,
                        raci_role, assigned_by
                    ) VALUES (
                        CAST(:id AS uuid), CAST(:scope_id AS uuid),
                        CAST(:user_id AS uuid), :responsibility_role,
                        :raci_role, CAST(:actor_uid AS uuid)
                    )
                    """
                ),
                {
                    "id": new_governance_uid(),
                    "scope_id": scope_id,
                    "user_id": assignment.user_id,
                    "responsibility_role": assignment.responsibility_role,
                    "raci_role": assignment.raci_role,
                    "actor_uid": actor_uid,
                },
            )

        new_revision = current_revision + 1
        self.session.execute(
            text(
                """
                UPDATE public.governance_responsibility_scopes
                SET revision = :revision, updated_by = CAST(:actor_uid AS uuid),
                    updated_at = CURRENT_TIMESTAMP
                WHERE id = CAST(:scope_id AS uuid)
                """
            ),
            {
                "scope_id": scope_id,
                "revision": new_revision,
                "actor_uid": actor_uid,
            },
        )
        requested = [asdict(item) for item in assignments]
        self.session.execute(
            text(
                """
                INSERT INTO public.governance_responsibility_audit_events (
                    scope_id, resource_type, resource_uid, actor_uid,
                    action, before_state, after_state
                ) VALUES (
                    CAST(:scope_id AS uuid), :resource_type, :resource_uid,
                    CAST(:actor_uid AS uuid), 'matrix_replaced',
                    CAST(:before_state AS jsonb), CAST(:after_state AS jsonb)
                )
                """
            ),
            {
                "scope_id": scope_id,
                "resource_type": resource_type,
                "resource_uid": resource_uid,
                "actor_uid": actor_uid,
                "before_state": json.dumps(before, ensure_ascii=False),
                "after_state": json.dumps(requested, ensure_ascii=False),
            },
        )
        return {
            "resource_type": resource_type,
            "resource_uid": resource_uid,
            "revision": new_revision,
            "assignments": self._assignments(scope_id),
        }